fix(cache): 优化仓库扫描、保留存储错误并收敛后台清理范围

internal/dao/cache_admin_benchmark_test.go

@@ -0,0 +1,88 @@
+package dao
+
+import (
+	"io/fs"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"testing"
+
+	"dingospeed/internal/data"
+	"dingospeed/pkg/config"
+)
+
+// BenchmarkListFilesExactRepo compares the current exact-key path with the old
+// call chain, which first discovered every repository using revision as the only
+// API marker. The fixture is intentionally small: seven repositories and 500
+// empty paths-info entries per repository.
+func BenchmarkListFilesExactRepo(b *testing.B) {
+	repos := b.TempDir()
+	oldConfig := config.SysConfig
+	config.SysConfig = &config.Config{
+		Server: config.ServerConfig{Repos: repos},
+		Upload: config.Upload{Namespace: "dingo-local"},
+	}
+	b.Cleanup(func() { config.SysConfig = oldConfig })
+
+	baseData := data.NewBaseData()
+	lockDao := NewLockDao(baseData)
+	admin := NewCacheAdminDao(NewFileDao(nil, baseData, lockDao))
+
+	for repo := 0; repo < 7; repo++ {
+		orgRepo := "remote/repo-" + string(rune('a'+repo))
+		if repo == 0 {
+			orgRepo = "dingo-local/target"
+		}
+		if err := os.MkdirAll(filepath.Join(repoApiRoot("models", orgRepo), "revision", "main"), 0o755); err != nil {
+			b.Fatal(err)
+		}
+		for file := 0; file < 500; file++ {
+			path := filepath.Join(repoApiRoot("models", orgRepo), "paths-info", "commit", "dir", strconv.Itoa(file))
+			if err := os.MkdirAll(path, 0o755); err != nil {
+				b.Fatal(err)
+			}
+		}
+	}
+
+	b.Run("direct", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			_ = admin.ListFiles("models", "dingo-local/target")
+		}
+	})
+	b.Run("legacy-global-discovery", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			_ = legacyListFilesForBenchmark("models", "dingo-local/target")
+		}
+	})
+}
+
+func legacyListFilesForBenchmark(repoType, orgRepo string) []*CacheFileRow {
+	keys := make(map[repoKey]struct{})
+	root := filepath.Join(config.SysConfig.Repos(), "api", repoType)
+	_ = filepath.WalkDir(root, func(path string, entry fs.DirEntry, err error) error {
+		if err != nil || !entry.IsDir() || path == root {
+			return nil
+		}
+		if entry.Name() != "revision" {
+			return nil
+		}
+		rel, relErr := filepath.Rel(root, filepath.Dir(path))
+		if relErr == nil && rel != "." && !strings.HasPrefix(rel, "..") {
+			keys[repoKey{RepoType: repoType, OrgRepo: filepath.ToSlash(rel)}] = struct{}{}
+		}
+		return fs.SkipDir
+	})
+	ordered := make([]repoKey, 0, len(keys))
+	for key := range keys {
+		ordered = append(ordered, key)
+	}
+	sort.Slice(ordered, func(i, j int) bool { return ordered[i].OrgRepo < ordered[j].OrgRepo })
+	for _, key := range ordered {
+		if key.OrgRepo == orgRepo {
+			return indexRows(buildRepoIndex(key.RepoType, key.OrgRepo))
+		}
+	}
+	return []*CacheFileRow{}
+}

internal/dao/cache_admin_dao.go

@@ -490,7 +490,9 @@ func listRepoKeys() []repoKey {
 	repos := config.SysConfig.Repos()
 	for _, repoType := range []string{"models", "datasets", "spaces"} {
 		scanRepoRoots(filepath.Join(repos, "files", repoType), repoType, []string{"blobs", "resolve"}, keys)
-		scanRepoRoots(filepath.Join(repos, "api", repoType), repoType, []string{"revision"}, keys)
+		// paths-info 和 recycle 都是仓库根下的数据子树。它们本身足以证明仓库
+		// 存在；命中后必须停止向下遍历，否则仓库发现会退化为扫描全部文件。
+		scanRepoRoots(filepath.Join(repos, "api", repoType), repoType, []string{"revision", "paths-info", recycleDirName}, keys)
 	}
 	result := make([]repoKey, 0, len(keys))
 	for k := range keys {
@@ -561,6 +563,16 @@ func (d *CacheAdminDao) ListRepos() []*CacheRepo {
 
 // ListFiles 返回某个仓库的一级列表；orgRepo 为空时返回全部仓库的合集。
 func (d *CacheAdminDao) ListFiles(repoType, orgRepo string) []*CacheFileRow {
+	// 调用方已经给出完整仓库键时直接构建该仓库的索引。不能为了确认它是否在
+	// 列表中先枚举所有仓库；一个小仓库的延迟不应受其他仓库大小影响。
+	if repoType != "" && orgRepo != "" {
+		key := repoKey{RepoType: repoType, OrgRepo: orgRepo}
+		if validateRepoKey(key) != nil {
+			return []*CacheFileRow{}
+		}
+		return indexRows(buildRepoIndex(repoType, orgRepo))
+	}
+
 	rows := make([]*CacheFileRow, 0)
 	for _, key := range listRepoKeys() {
 		if repoType != "" && key.RepoType != repoType {

internal/dao/cache_admin_dao_test.go

@@ -115,6 +115,26 @@ func findOrphan(rows []*RecycleRow, sha string) *RecycleRow {
 	return nil
 }
 
+func TestListRepoKeysStopsAtPathsInfo(t *testing.T) {
+	_, _, _ = newTestCacheAdminDao(t)
+	orgRepo := "dingo-local/paths-heavy"
+
+	// revision 是一个合法仓库标记，但这里故意把同名目录放在 paths-info
+	// 深处。仓库发现如果进入数据子树，会把这个深层路径误判成另一个仓库。
+	deepRevision := filepath.Join(repoApiRoot("models", orgRepo), "paths-info", "commit", "nested", "revision")
+	if err := os.MkdirAll(deepRevision, 0o755); err != nil {
+		t.Fatalf("create deep paths-info fixture: %v", err)
+	}
+
+	keys := listRepoKeys()
+	if len(keys) != 1 {
+		t.Fatalf("expected one repository without descending paths-info, got %#v", keys)
+	}
+	if keys[0] != (repoKey{RepoType: "models", OrgRepo: orgRepo}) {
+		t.Fatalf("unexpected repository key: %#v", keys[0])
+	}
+}
+
 func tombstoneExists(repoType, orgRepo, sha string) bool {
 	return util.FileExists(recycleEntryPath(repoType, orgRepo, sha))
 }

internal/dao/file_dao.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -82,6 +83,11 @@ func (f *FileDao) GetFileCommitSha(repoType, orgRepo, commit, authorization stri
 	if IsLocalOrgRepo(orgRepo) {
 		commitSha, err = f.GetCommitHfOffline(repoType, orgRepo, commit)
 		if err != nil {
+			var accessErr *util.FileAccessError
+			if errors.As(err, &accessErr) {
+				zap.S().Errorw("local metadata storage unavailable", "kind", accessErr.Kind, "path", accessErr.Path, "error", accessErr.Err)
+				return "", myerr.Wrap("local metadata storage unavailable", err)
+			}
 			return "", myerr.NewAppendCode(http.StatusNotFound, fmt.Sprintf("%s is not found", orgRepo))
 		}
 		return commitSha, nil
@@ -91,6 +97,11 @@ func (f *FileDao) GetFileCommitSha(repoType, orgRepo, commit, authorization stri
 	}
 	commitSha, err = f.GetCommitHfOffline(repoType, orgRepo, commit)
 	if err != nil {
+		var accessErr *util.FileAccessError
+		if errors.As(err, &accessErr) {
+			zap.S().Errorw("metadata cache storage unavailable", "kind", accessErr.Kind, "path", accessErr.Path, "error", accessErr.Err)
+			return "", myerr.Wrap("metadata cache storage unavailable", err)
+		}
 		if source == "file" {
 			// 若只是发起文件下载（先在线后离线），将不会校验meta文件是否存在，没有就创建，主要是看文件本身是否存在。
 			goto remoteRequestMeta
@@ -167,9 +178,16 @@ func (f *FileDao) RemoteRequestMeta(method, repoType, orgRepo, revision, authori
 
 func (f *FileDao) GetCommitHfOffline(repoType, orgRepo, commit string) (string, error) {
 	apiPath := fmt.Sprintf("%s/api/%s/%s/revision/%s/meta_get.json", config.SysConfig.Repos(), repoType, orgRepo, commit)
-	if util.FileExists(apiPath) {
+	exists, err := util.PathExists(apiPath)
+	if err != nil {
+		return "", err
+	}
+	if exists {
 		cacheContent, err := f.ReadCacheRequest(apiPath)
 		if err != nil {
+			if accessErr, ok := util.ClassifyFileAccessError(apiPath, err); ok {
+				return "", accessErr
+			}
 			return "", err
 		}
 		var sha CommitHfSha

internal/dao/file_dao_storage_error_test.go

@@ -0,0 +1,50 @@
+package dao
+
+import (
+	"errors"
+	"net/http"
+	"strings"
+	"testing"
+
+	"dingospeed/internal/data"
+	"dingospeed/pkg/config"
+	myerr "dingospeed/pkg/error"
+	"dingospeed/pkg/util"
+)
+
+func TestGetCommitHfOfflinePreservesMissingBehavior(t *testing.T) {
+	oldConfig := config.SysConfig
+	config.SysConfig = &config.Config{Server: config.ServerConfig{Repos: t.TempDir()}}
+	t.Cleanup(func() { config.SysConfig = oldConfig })
+
+	fileDao := NewFileDao(nil, data.NewBaseData(), nil)
+	_, err := fileDao.GetFileCommitSha("models", "org/repo", "main", "", "meta")
+	var appErr myerr.Error
+	if !errors.As(err, &appErr) || appErr.StatusCode() != http.StatusNotFound {
+		t.Fatalf("missing metadata: got %v, want HTTP 404 application error", err)
+	}
+}
+
+func TestGetFileCommitShaPreservesStorageFailure(t *testing.T) {
+	oldConfig := config.SysConfig
+	// A NUL byte makes stat fail as an invalid path on every supported OS. It
+	// exercises the inaccessible-path branch without relying on host mounts.
+	config.SysConfig = &config.Config{Server: config.ServerConfig{Repos: "invalid\x00repos"}}
+	t.Cleanup(func() { config.SysConfig = oldConfig })
+
+	fileDao := NewFileDao(nil, data.NewBaseData(), nil)
+	_, err := fileDao.GetFileCommitSha("models", "org/repo", "main", "", "meta")
+	if err == nil {
+		t.Fatal("expected inaccessible storage error")
+	}
+	var accessErr *util.FileAccessError
+	if !errors.As(err, &accessErr) {
+		t.Fatalf("got %T %v, want wrapped FileAccessError", err, err)
+	}
+	if accessErr.Kind != util.FileAccessUnavailable {
+		t.Fatalf("got kind %q, want %q", accessErr.Kind, util.FileAccessUnavailable)
+	}
+	if strings.Contains(err.Error(), "invalid\x00repos") {
+		t.Fatalf("public application error exposes storage path: %q", err.Error())
+	}
+}

internal/dao/meta_dao.go

@@ -15,6 +15,7 @@
 package dao
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"path/filepath"
@@ -191,22 +192,22 @@ func (m *MetaDao) requestAndSaveMeta(repoType, orgRepo, revision, commitSha, met
 	if revision == mainVersion {
 		err = m.writeApiMetaFile(repoType, orgRepo, revision, method, resp.StatusCode, extractHeaders, resp.Body)
 		if err != nil {
-			return nil, err
+			m.logMetadataCacheWriteFailure(orgRepo, revision, method, err)
 		}
 	} else {
 		apiDir := fmt.Sprintf("%s/api/%s/%s/revision/%s", config.SysConfig.Repos(), repoType, orgRepo, mainVersion)
 		apiMetaPath := fmt.Sprintf("%s/%s", apiDir, fmt.Sprintf("meta_%s.json", method))
 		if !util.FileExists(apiMetaPath) {
 			err = m.writeApiMetaFile(repoType, orgRepo, mainVersion, method, resp.StatusCode, extractHeaders, resp.Body) // create main dir
 			if err != nil {
-				return nil, err
+				m.logMetadataCacheWriteFailure(orgRepo, mainVersion, method, err)
 			}
 		}
 	}
 
 	err = m.writeApiMetaFile(repoType, orgRepo, commitSha, method, resp.StatusCode, extractHeaders, resp.Body)
 	if err != nil {
-		return nil, err
+		m.logMetadataCacheWriteFailure(orgRepo, commitSha, method, err)
 	}
 	return &common.CacheContent{
 		StatusCode:    resp.StatusCode,
@@ -215,16 +216,31 @@ func (m *MetaDao) requestAndSaveMeta(repoType, orgRepo, revision, commitSha, met
 	}, nil
 }
 
+func (m *MetaDao) logMetadataCacheWriteFailure(orgRepo, revision, method string, err error) {
+	fields := []interface{}{"repo", orgRepo, "revision", revision, "method", method, "error", err}
+	var accessErr *util.FileAccessError
+	if errors.As(err, &accessErr) {
+		fields = append(fields, "kind", accessErr.Kind, "path", accessErr.Path)
+	}
+	zap.S().Warnw("serving remote metadata without cache", fields...)
+}
+
 func (m *MetaDao) writeApiMetaFile(repoType, orgRepo, commitSha, method string, statusCode int, extractHeaders map[string]string, body []byte) error {
 	apiDir := fmt.Sprintf("%s/api/%s/%s/revision/%s", config.SysConfig.Repos(), repoType, orgRepo, commitSha)
 	apiMetaPath := fmt.Sprintf("%s/%s", apiDir, fmt.Sprintf("meta_%s.json", method))
 	err := util.MakeDirs(apiMetaPath)
 	if err != nil {
 		zap.S().Errorf("create %s dir err.%v", apiMetaPath, err)
+		if accessErr, ok := util.ClassifyFileAccessError(apiMetaPath, err); ok {
+			return accessErr
+		}
 		return err
 	}
 	if err = m.fileDao.WriteCacheRequest(apiMetaPath, statusCode, extractHeaders, body); err != nil {
 		zap.S().Errorf("writeCacheRequest err.%v", err)
+		if accessErr, ok := util.ClassifyFileAccessError(apiMetaPath, err); ok {
+			return accessErr
+		}
 		return err
 	}
 	return nil

internal/dao/meta_dao_degraded_test.go

@@ -0,0 +1,80 @@
+package dao
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"dingospeed/internal/data"
+	"dingospeed/pkg/config"
+	"dingospeed/pkg/consts"
+)
+
+func TestRequestAndSaveMetaServesRemoteResponseWhenCacheUnavailable(t *testing.T) {
+	const body = `{"sha":"remote-commit","id":"org/repo"}`
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(body))
+	}))
+	t.Cleanup(server.Close)
+
+	oldConfig := config.SysConfig
+	config.SysConfig = &config.Config{
+		Server: config.ServerConfig{
+			Online:   true,
+			Repos:    "invalid\x00repos",
+			HfScheme: "http",
+			HfNetLoc: strings.TrimPrefix(server.URL, "http://"),
+		},
+		Retry: config.Retry{Attempts: 1},
+	}
+	t.Cleanup(func() { config.SysConfig = oldConfig })
+
+	baseData := data.NewBaseData()
+	fileDao := NewFileDao(nil, baseData, NewLockDao(baseData))
+	metaDao := NewMetaDao(fileDao, nil, baseData)
+	got, err := metaDao.requestAndSaveMeta("models", "org/repo", "main", "remote-commit", consts.RequestTypeGet, "")
+	if err != nil {
+		t.Fatalf("remote metadata should survive cache failure: %v", err)
+	}
+	if string(got.OriginContent) != body {
+		t.Fatalf("got body %q, want %q", got.OriginContent, body)
+	}
+}
+
+func TestRequestAndSaveMetaStillCachesWhenStorageAvailable(t *testing.T) {
+	const body = `{"sha":"remote-commit","id":"org/repo"}`
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(body))
+	}))
+	t.Cleanup(server.Close)
+
+	repos := t.TempDir()
+	oldConfig := config.SysConfig
+	config.SysConfig = &config.Config{
+		Server: config.ServerConfig{
+			Online:   true,
+			Repos:    repos,
+			HfScheme: "http",
+			HfNetLoc: strings.TrimPrefix(server.URL, "http://"),
+		},
+		Retry: config.Retry{Attempts: 1},
+	}
+	t.Cleanup(func() { config.SysConfig = oldConfig })
+
+	baseData := data.NewBaseData()
+	fileDao := NewFileDao(nil, baseData, NewLockDao(baseData))
+	metaDao := NewMetaDao(fileDao, nil, baseData)
+	if _, err := metaDao.requestAndSaveMeta("models", "org/repo", "main", "remote-commit", consts.RequestTypeGet, ""); err != nil {
+		t.Fatal(err)
+	}
+	for _, revision := range []string{"main", "remote-commit"} {
+		path := filepath.Join(repos, "api", "models", "org", "repo", "revision", revision, "meta_get.json")
+		if _, err := os.Stat(path); err != nil {
+			t.Fatalf("expected metadata cache %s: %v", path, err)
+		}
+	}
+}