diff --git a/internal/repo/file_record/file_record_repo.go b/internal/repo/file_record/file_record_repo.go index ce486c7ab..b305effb2 100644 --- a/internal/repo/file_record/file_record_repo.go +++ b/internal/repo/file_record/file_record_repo.go @@ -22,12 +22,10 @@ package file_record import ( "context" - "github.com/apache/answer/internal/base/pager" - "github.com/apache/answer/internal/service/file_record" - "github.com/apache/answer/internal/base/data" "github.com/apache/answer/internal/base/reason" "github.com/apache/answer/internal/entity" + "github.com/apache/answer/internal/service/file_record" "github.com/segmentfault/pacman/errors" ) @@ -52,13 +50,20 @@ func (fr *fileRecordRepo) AddFileRecord(ctx context.Context, fileRecord *entity. return } -// GetFileRecordPage get fileRecord page -func (fr *fileRecordRepo) GetFileRecordPage(ctx context.Context, page, pageSize int, cond *entity.FileRecord) ( - fileRecordList []*entity.FileRecord, total int64, err error) { +// GetFileRecordListAfterID get available file record list whose id is greater than lastID, +// the result is ordered by id ascending. It is used to scan all available file records by +// id cursor, because the scanned records may be marked as deleted during the scan and +// offset pagination would skip records in that case. +func (fr *fileRecordRepo) GetFileRecordListAfterID(ctx context.Context, lastID, limit int) ( + fileRecordList []*entity.FileRecord, err error) { fileRecordList = make([]*entity.FileRecord, 0) - session := fr.data.DB.Context(ctx) - total, err = pager.Help(page, pageSize, &fileRecordList, cond, session) + err = fr.data.DB.Context(ctx). + Where("status = ?", entity.FileRecordStatusAvailable). + And("id > ?", lastID). + OrderBy("id ASC"). + Limit(limit). + Find(&fileRecordList) if err != nil { err = errors.InternalServer(reason.DatabaseError).WithError(err).WithStack() } diff --git a/internal/service/file_record/file_record_service.go b/internal/service/file_record/file_record_service.go index aa526f014..f52eba142 100644 --- a/internal/service/file_record/file_record_service.go +++ b/internal/service/file_record/file_record_service.go @@ -43,8 +43,7 @@ import ( type FileRecordRepo interface { AddFileRecord(ctx context.Context, fileRecord *entity.FileRecord) (err error) UpdateFileRecord(ctx context.Context, fileRecord *entity.FileRecord) (err error) - GetFileRecordPage(ctx context.Context, page, pageSize int, cond *entity.FileRecord) ( - fileRecordList []*entity.FileRecord, total int64, err error) + GetFileRecordListAfterID(ctx context.Context, lastID, limit int) (fileRecordList []*entity.FileRecord, err error) DeleteFileRecord(ctx context.Context, id int) (err error) GetFileRecordByURL(ctx context.Context, fileURL string) (record *entity.FileRecord, err error) } @@ -92,20 +91,23 @@ func (fs *FileRecordService) AddFileRecord(ctx context.Context, userID, filePath // CleanOrphanUploadFiles clean orphan upload files func (fs *FileRecordService) CleanOrphanUploadFiles(ctx context.Context) { - page, pageSize := 1, 1000 + pageSize := 1000 + // Scan available file records by id cursor. Records are marked as deleted during the + // scan, so they leave the available result set: offset pagination would shift the + // remaining records forward and skip some of them. + lastID := 0 for { - fileRecordList, total, err := fs.fileRecordRepo.GetFileRecordPage(ctx, page, pageSize, &entity.FileRecord{ - Status: entity.FileRecordStatusAvailable, - }) + fileRecordList, err := fs.fileRecordRepo.GetFileRecordListAfterID(ctx, lastID, pageSize) if err != nil { - log.Errorf("get file record page error: %v", err) + log.Errorf("get file record list error: %v", err) return } - if len(fileRecordList) == 0 || total == 0 { + if len(fileRecordList) == 0 { break } for _, fileRecord := range fileRecordList { + lastID = fileRecord.ID // If this file record created in 48 hours, no need to check if fileRecord.CreatedAt.AddDate(0, 0, 2).After(time.Now()) { continue @@ -154,7 +156,6 @@ func (fs *FileRecordService) CleanOrphanUploadFiles(ctx context.Context) { log.Error(err) } } - page++ } } diff --git a/internal/service/file_record/file_record_service_test.go b/internal/service/file_record/file_record_service_test.go new file mode 100644 index 000000000..85bb5f004 --- /dev/null +++ b/internal/service/file_record/file_record_service_test.go @@ -0,0 +1,205 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package file_record + +import ( + "context" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/apache/answer/internal/base/constant" + "github.com/apache/answer/internal/entity" + "github.com/apache/answer/internal/service/service_config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "xorm.io/xorm" +) + +// fakeFileRecordRepo is an in-memory file record repo. It keeps the same semantics as the +// database one: GetFileRecordListAfterID only returns available records ordered by id, and +// DeleteFileRecord marks a record as deleted so that it leaves the available result set. +type fakeFileRecordRepo struct { + records []*entity.FileRecord + listCalls int +} + +func (f *fakeFileRecordRepo) AddFileRecord(_ context.Context, fileRecord *entity.FileRecord) error { + f.records = append(f.records, fileRecord) + return nil +} + +func (f *fakeFileRecordRepo) UpdateFileRecord(_ context.Context, fileRecord *entity.FileRecord) error { + return nil +} + +func (f *fakeFileRecordRepo) GetFileRecordListAfterID(_ context.Context, lastID, limit int) ( + []*entity.FileRecord, error) { + f.listCalls++ + list := make([]*entity.FileRecord, 0) + for _, record := range f.records { + if record.Status != entity.FileRecordStatusAvailable || record.ID <= lastID { + continue + } + list = append(list, record) + if len(list) == limit { + break + } + } + return list, nil +} + +func (f *fakeFileRecordRepo) DeleteFileRecord(_ context.Context, id int) error { + for _, record := range f.records { + if record.ID == id { + record.Status = entity.FileRecordStatusDeleted + return nil + } + } + return fmt.Errorf("file record %d not found", id) +} + +func (f *fakeFileRecordRepo) GetFileRecordByURL(_ context.Context, fileURL string) ( + *entity.FileRecord, error) { + for _, record := range f.records { + if record.FileURL == fileURL && record.Status == entity.FileRecordStatusAvailable { + return record, nil + } + } + return nil, nil +} + +// fakeRevisionRepo is a revision repo stub which reports the given file urls as referenced. +type fakeRevisionRepo struct { + usedFileURLs map[string]*entity.Revision +} + +func (f *fakeRevisionRepo) AddRevision(_ context.Context, _ *entity.Revision, _ bool) error { + return nil +} + +func (f *fakeRevisionRepo) GetRevisionByID(_ context.Context, _ string) (*entity.Revision, bool, error) { + return nil, false, nil +} + +func (f *fakeRevisionRepo) GetLastRevisionByObjectID(_ context.Context, _ string) (*entity.Revision, bool, error) { + return nil, false, nil +} + +func (f *fakeRevisionRepo) GetLastRevisionByFileURL(_ context.Context, fileURL string) (*entity.Revision, bool, error) { + if revision, ok := f.usedFileURLs[fileURL]; ok { + return revision, true, nil + } + return nil, false, nil +} + +func (f *fakeRevisionRepo) GetRevisionList(_ context.Context, _ *entity.Revision) ([]entity.Revision, error) { + return nil, nil +} + +func (f *fakeRevisionRepo) UpdateObjectRevisionId(_ context.Context, _ *entity.Revision, _ *xorm.Session) error { + return nil +} + +func (f *fakeRevisionRepo) ExistUnreviewedByObjectID(_ context.Context, _ string) (*entity.Revision, bool, error) { + return nil, false, nil +} + +func (f *fakeRevisionRepo) GetUnreviewedRevisionPage(_ context.Context, _, _ int, _ []int) ( + []*entity.Revision, int64, error) { + return nil, 0, nil +} + +func (f *fakeRevisionRepo) CountUnreviewedRevision(_ context.Context, _ []int) (int64, error) { + return 0, nil +} + +func (f *fakeRevisionRepo) UpdateStatus(_ context.Context, _ string, _ int, _ string) error { + return nil +} + +// TestFileRecordService_CleanOrphanUploadFiles checks that all orphan files are cleaned even +// though the scan marks records as deleted while iterating, which used to make offset +// pagination skip records between pages. +func TestFileRecordService_CleanOrphanUploadFiles(t *testing.T) { + const ( + recordCount = 2500 + referencedRecordID = 1234 + recentRecordID = 2500 + ) + + uploadPath := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(uploadPath, constant.DeletedSubPath), 0o755)) + require.NoError(t, os.MkdirAll(filepath.Join(uploadPath, "files", "post"), 0o755)) + + repo := &fakeFileRecordRepo{} + revisionRepo := &fakeRevisionRepo{usedFileURLs: map[string]*entity.Revision{}} + oldCreatedAt := time.Now().Add(-time.Hour * 72) + for id := 1; id <= recordCount; id++ { + filePath := fmt.Sprintf("files/post/%d.txt", id) + fileURL := fmt.Sprintf("uploads/%s", filePath) + require.NoError(t, os.WriteFile(filepath.Join(uploadPath, filePath), []byte("x"), 0o644)) + + createdAt := oldCreatedAt + if id == recentRecordID { + createdAt = time.Now() + } + repo.records = append(repo.records, &entity.FileRecord{ + ID: id, + CreatedAt: createdAt, + FilePath: filePath, + FileURL: fileURL, + ObjectID: "0", + Status: entity.FileRecordStatusAvailable, + }) + if id == referencedRecordID { + revisionRepo.usedFileURLs[fileURL] = &entity.Revision{ObjectID: "100"} + } + } + + serviceConfig := &service_config.ServiceConfig{UploadPath: uploadPath} + service := NewFileRecordService(repo, revisionRepo, serviceConfig, nil, nil) + + service.CleanOrphanUploadFiles(context.Background()) + + assert.GreaterOrEqual(t, repo.listCalls, 3, "all records should be scanned page by page") + + deletedCount := 0 + for _, record := range repo.records { + filePath := filepath.Join(uploadPath, record.FilePath) + deletedFilePath := filepath.Join(uploadPath, constant.DeletedSubPath, filepath.Base(record.FilePath)) + switch record.ID { + case referencedRecordID, recentRecordID: + assert.Equal(t, entity.FileRecordStatusAvailable, record.Status, + "record %d should not be deleted", record.ID) + assert.FileExists(t, filePath, "file of record %d should stay in place", record.ID) + assert.NoFileExists(t, deletedFilePath, "file of record %d should not be moved", record.ID) + default: + deletedCount++ + assert.Equal(t, entity.FileRecordStatusDeleted, record.Status, + "record %d should be deleted", record.ID) + assert.NoFileExists(t, filePath, "file of record %d should be moved away", record.ID) + assert.FileExists(t, deletedFilePath, "file of record %d should be moved to deleted dir", record.ID) + } + } + assert.Equal(t, recordCount-2, deletedCount) +}