feat: hash-based duplicate detection and instant upload (#191)
* feat: hash-based duplicate detection and instant upload Clients send a client-computed sha256 with the upload session request; when upload_dedup_scope is enabled and a completed entity with the same hash and size exists within scope, the file is materialized by linking the existing entity (refcount++) instead of transferring bytes. - entity.hash column + (hash,size) index; stored on entity creation - FileClient.FindEntityByHash: completed version entities only (reference_count > 0, no upload session), owner-scoped unless global - CreateFile LinkedEntityID branch: attach entity, bump refcount, set primary entity + logical size, charge owner quota via storage diff - DBFS.PrepareUpload rapid path: dedup lookup after validation, commits file + emits upload event, skips session metadata/lock/credentials - manager short-circuits rapid sessions before driver token + KV store - postProcessRapidUpload queues media-meta / FTS tasks per file - upload_dedup_scope setting (off/owner/global, default owner) surfaced in admin FileSystem parameters and basic site config - Frontend streams SHA-256 via hash-wasm (no whole-file buffering), skips hashing for encrypted policies, short-circuits on rapid_uploaded - Tests: FindEntityByHash scoping/isolation, linked CreateFile, and full PrepareUpload rapid path incl. scope=off Implements upstream cloudreve/cloudreve#3044 (dedup detection); addresses fork issues #66/#15. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * docs: roadmap — dedup landed (#191), multi-file share scoped --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>pull/3587/head
parent
5e7f938f41
commit
cf688634c4
File diff suppressed because one or more lines are too long
@ -0,0 +1,143 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/cloudreve/Cloudreve/v4/ent"
|
||||
"github.com/cloudreve/Cloudreve/v4/ent/enttest"
|
||||
"github.com/cloudreve/Cloudreve/v4/ent/storagepolicy"
|
||||
"github.com/cloudreve/Cloudreve/v4/inventory/types"
|
||||
"github.com/cloudreve/Cloudreve/v4/pkg/boolset"
|
||||
"github.com/cloudreve/Cloudreve/v4/pkg/conf"
|
||||
"github.com/gofrs/uuid"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func dedupFixture(t *testing.T, client *ent.Client, suffix string) (*ent.User, *ent.File, *ent.StoragePolicy) {
|
||||
ctx := context.Background()
|
||||
group := client.Group.Create().SetName("g" + suffix).SetPermissions(&boolset.BooleanSet{}).SaveX(ctx)
|
||||
u := client.User.Create().SetEmail("u" + suffix + "@example.com").SetNick("u" + suffix).SetGroup(group).SaveX(ctx)
|
||||
root := client.File.Create().SetName(RootFolderName).SetType(int(types.FileTypeFolder)).SetOwner(u).SaveX(ctx)
|
||||
p := client.StoragePolicy.Create().SetName("p" + suffix).SetType("local").
|
||||
SetStatus(storagepolicy.StatusActive).SaveX(ctx)
|
||||
return u, root, p
|
||||
}
|
||||
|
||||
func mkHashedEntity(client *ent.Client, ctx context.Context, u *ent.User, p *ent.StoragePolicy, hash string, size int64) *ent.Entity {
|
||||
return client.Entity.Create().
|
||||
SetType(int(types.EntityTypeVersion)).
|
||||
SetSource("cloudreve/" + hash).
|
||||
SetSize(size).
|
||||
SetHash(hash).
|
||||
SetReferenceCount(1).
|
||||
SetCreatedBy(u.ID).
|
||||
SetStoragePolicyEntities(p.ID).
|
||||
SaveX(ctx)
|
||||
}
|
||||
|
||||
func TestFindEntityByHash(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
client := enttest.Open(t, "sqlite3", "file:"+t.Name()+"?mode=memory&cache=shared")
|
||||
t.Cleanup(func() { require.NoError(t, client.Close()) })
|
||||
fc := NewFileClient(client, conf.SQLite3DB, nil)
|
||||
|
||||
u, _, p := dedupFixture(t, client, "a")
|
||||
other, _, _ := dedupFixture(t, client, "b")
|
||||
|
||||
const hash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
|
||||
e := mkHashedEntity(client, ctx, u, p, hash, 100)
|
||||
|
||||
// Owner scope: match
|
||||
found, err := fc.FindEntityByHash(ctx, hash, 100, u.ID, false)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, e.ID, found.ID)
|
||||
|
||||
// Owner scope: other user's entity must not leak
|
||||
_, err = fc.FindEntityByHash(ctx, hash, 100, other.ID, false)
|
||||
require.True(t, ent.IsNotFound(err))
|
||||
|
||||
// Global scope: cross-user match allowed
|
||||
found, err = fc.FindEntityByHash(ctx, hash, 100, other.ID, true)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, e.ID, found.ID)
|
||||
|
||||
// Size mismatch: no match
|
||||
_, err = fc.FindEntityByHash(ctx, hash, 101, u.ID, false)
|
||||
require.True(t, ent.IsNotFound(err))
|
||||
|
||||
// Empty hash: no match
|
||||
_, err = fc.FindEntityByHash(ctx, "", 100, u.ID, false)
|
||||
require.True(t, ent.IsNotFound(err))
|
||||
|
||||
// Unfinished upload entity (session still attached): excluded
|
||||
unfinished := client.Entity.Create().
|
||||
SetType(int(types.EntityTypeVersion)).
|
||||
SetSource("cloudreve/unfinished").
|
||||
SetSize(50).
|
||||
SetHash("aaaa").
|
||||
SetReferenceCount(0).
|
||||
SetCreatedBy(u.ID).
|
||||
SetUploadSessionID(uuid.Must(uuid.NewV4())).
|
||||
SetStoragePolicyEntities(p.ID).
|
||||
SaveX(ctx)
|
||||
_, err = fc.FindEntityByHash(ctx, "aaaa", 50, u.ID, false)
|
||||
require.True(t, ent.IsNotFound(err))
|
||||
_ = unfinished
|
||||
|
||||
// Stale entity (refcount 0, session cleared): excluded
|
||||
client.Entity.Create().
|
||||
SetType(int(types.EntityTypeVersion)).
|
||||
SetSource("cloudreve/stale").
|
||||
SetSize(60).
|
||||
SetHash("bbbb").
|
||||
SetReferenceCount(0).
|
||||
SetCreatedBy(u.ID).
|
||||
SetStoragePolicyEntities(p.ID).
|
||||
SaveX(ctx)
|
||||
_, err = fc.FindEntityByHash(ctx, "bbbb", 60, u.ID, false)
|
||||
require.True(t, ent.IsNotFound(err))
|
||||
}
|
||||
|
||||
func TestCreateFileLinkedEntity(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
client := enttest.Open(t, "sqlite3", "file:"+t.Name()+"?mode=memory&cache=shared")
|
||||
t.Cleanup(func() { require.NoError(t, client.Close()) })
|
||||
fc := NewFileClient(client, conf.SQLite3DB, nil)
|
||||
|
||||
u, root, p := dedupFixture(t, client, "l")
|
||||
|
||||
const hash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
|
||||
existing := mkHashedEntity(client, ctx, u, p, hash, 512)
|
||||
|
||||
newFile, linked, diff, err := fc.CreateFile(ctx, root, &CreateFileParameters{
|
||||
FileType: types.FileTypeFile,
|
||||
Name: "copy.txt",
|
||||
StoragePolicyID: p.ID,
|
||||
EntityParameters: &EntityParameters{
|
||||
LinkedEntityID: existing.ID,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, existing.ID, linked.ID)
|
||||
|
||||
// Refcount incremented
|
||||
reloaded := client.Entity.GetX(ctx, existing.ID)
|
||||
require.Equal(t, 2, reloaded.ReferenceCount)
|
||||
|
||||
// File wired to the shared entity (reload: primary/size are set via
|
||||
// a follow-up update after the initial insert)
|
||||
reloadedFile := client.File.GetX(ctx, newFile.ID)
|
||||
require.Equal(t, int64(512), reloadedFile.Size)
|
||||
require.Equal(t, existing.ID, reloadedFile.PrimaryEntity)
|
||||
require.Equal(t, u.ID, reloadedFile.OwnerID)
|
||||
|
||||
// Logical quota charged to the new file's owner
|
||||
require.Equal(t, int64(512), diff[u.ID])
|
||||
|
||||
// Entity->file edge attached
|
||||
require.Len(t, reloaded.Edges.File, 0) // edges not eager-loaded
|
||||
files := reloaded.QueryFile().AllX(ctx)
|
||||
require.Len(t, files, 1)
|
||||
require.Equal(t, newFile.ID, files[0].ID)
|
||||
}
|
||||
@ -0,0 +1,150 @@
|
||||
package dbfs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/cloudreve/Cloudreve/v4/ent"
|
||||
"github.com/cloudreve/Cloudreve/v4/ent/enttest"
|
||||
entfile "github.com/cloudreve/Cloudreve/v4/ent/file"
|
||||
"github.com/cloudreve/Cloudreve/v4/ent/storagepolicy"
|
||||
"github.com/cloudreve/Cloudreve/v4/inventory"
|
||||
"github.com/cloudreve/Cloudreve/v4/inventory/types"
|
||||
"github.com/cloudreve/Cloudreve/v4/pkg/boolset"
|
||||
"github.com/cloudreve/Cloudreve/v4/pkg/conf"
|
||||
"github.com/cloudreve/Cloudreve/v4/pkg/filemanager/eventhub"
|
||||
"github.com/cloudreve/Cloudreve/v4/pkg/filemanager/fs"
|
||||
"github.com/cloudreve/Cloudreve/v4/pkg/filemanager/lock"
|
||||
"github.com/cloudreve/Cloudreve/v4/pkg/hashid"
|
||||
"github.com/cloudreve/Cloudreve/v4/pkg/logging"
|
||||
"github.com/cloudreve/Cloudreve/v4/pkg/setting"
|
||||
"github.com/gofrs/uuid"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type dedupSettingProvider struct {
|
||||
setting.Provider
|
||||
scope string
|
||||
}
|
||||
|
||||
func (p dedupSettingProvider) DBFS(context.Context) *setting.DBFS {
|
||||
return &setting.DBFS{DedupScope: p.scope, MaxPageSize: 200}
|
||||
}
|
||||
|
||||
type stubEventHub struct{}
|
||||
|
||||
func (stubEventHub) Subscribe(context.Context, int, string) (chan *eventhub.Event, bool, error) {
|
||||
return nil, false, nil
|
||||
}
|
||||
func (stubEventHub) Unsubscribe(context.Context, int, string) {}
|
||||
func (stubEventHub) GetSubscribers(context.Context, int) []eventhub.Subscriber {
|
||||
return nil
|
||||
}
|
||||
func (stubEventHub) Close() {}
|
||||
|
||||
func dedupUploadFixture(t *testing.T, client *ent.Client, scope string) (*ent.User, *ent.StoragePolicy, *DBFS) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
l := logging.NewConsoleLogger(logging.LevelError)
|
||||
hasher, err := hashid.New("dedup-test-salt")
|
||||
require.NoError(t, err)
|
||||
|
||||
p := client.StoragePolicy.Create().SetName("local").SetType("local").
|
||||
SetStatus(storagepolicy.StatusActive).SetSettings(&types.PolicySetting{}).SaveX(ctx)
|
||||
group := client.Group.Create().SetName("g").SetPermissions(&boolset.BooleanSet{}).
|
||||
SetMaxStorage(1 << 40).SetStoragePolicies(p).SaveX(ctx)
|
||||
u := client.User.Create().SetEmail("u@example.com").SetNick("u").SetGroup(group).SaveX(ctx)
|
||||
u.SetGroup(group)
|
||||
client.File.Create().SetName(inventory.RootFolderName).
|
||||
SetType(int(types.FileTypeFolder)).SetOwner(u).SaveX(ctx)
|
||||
|
||||
f := &DBFS{
|
||||
user: u,
|
||||
navigators: make(map[string]Navigator),
|
||||
fileClient: inventory.NewFileClient(client, conf.SQLiteDB, hasher),
|
||||
userClient: inventory.NewUserClient(client),
|
||||
storagePolicyClient: inventory.NewStoragePolicyClient(client, nil),
|
||||
settingClient: dedupSettingProvider{scope: scope},
|
||||
hasher: hasher,
|
||||
l: l,
|
||||
ls: lock.NewMemLS(hasher, l),
|
||||
eventHub: stubEventHub{},
|
||||
}
|
||||
return u, p, f
|
||||
}
|
||||
|
||||
const dedupTestHash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
|
||||
|
||||
func uploadReq(t *testing.T, hasher hashid.Encoder, u *ent.User, name string, size int64, hash string) *fs.UploadRequest {
|
||||
t.Helper()
|
||||
uri, err := fs.NewUriFromString(fs.NewMyUri(hashid.EncodeUserID(hasher, u.ID)) + "/" + name)
|
||||
require.NoError(t, err)
|
||||
return &fs.UploadRequest{
|
||||
Props: &fs.UploadProps{
|
||||
Uri: uri,
|
||||
Size: size,
|
||||
Hash: hash,
|
||||
UploadSessionID: uuid.Must(uuid.NewV4()).String(),
|
||||
ExpireAt: time.Now().Add(time.Hour),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func seedCompletedEntity(t *testing.T, client *ent.Client, u *ent.User, p *ent.StoragePolicy, hash string, size int64) *ent.Entity {
|
||||
t.Helper()
|
||||
return client.Entity.Create().
|
||||
SetType(int(types.EntityTypeVersion)).
|
||||
SetSource("cloudreve/data/" + hash).
|
||||
SetSize(size).
|
||||
SetHash(hash).
|
||||
SetReferenceCount(1).
|
||||
SetCreatedBy(u.ID).
|
||||
SetStoragePolicyEntities(p.ID).
|
||||
SaveX(context.Background())
|
||||
}
|
||||
|
||||
func TestPrepareUploadRapid(t *testing.T) {
|
||||
client := enttest.Open(t, "sqlite3", "file:"+t.Name()+"?mode=memory&cache=shared")
|
||||
t.Cleanup(func() { require.NoError(t, client.Close()) })
|
||||
ctx := context.Background()
|
||||
|
||||
u, p, f := dedupUploadFixture(t, client, "owner")
|
||||
existing := seedCompletedEntity(t, client, u, p, dedupTestHash, 1024)
|
||||
|
||||
// Normal upload without hash -> transfer session
|
||||
s, err := f.PrepareUpload(ctx, uploadReq(t, f.hasher, u, "first.txt", 2048, ""))
|
||||
require.NoError(t, err)
|
||||
require.False(t, s.Props.RapidUploaded)
|
||||
require.NotEmpty(t, s.Props.UploadSessionID)
|
||||
|
||||
// Same hash + size -> rapid session, no transfer required
|
||||
s, err = f.PrepareUpload(ctx, uploadReq(t, f.hasher, u, "copy.txt", 1024, dedupTestHash))
|
||||
require.NoError(t, err)
|
||||
require.True(t, s.Props.RapidUploaded)
|
||||
require.Equal(t, existing.ID, s.EntityID)
|
||||
|
||||
// Entity refcount bumped, new file linked
|
||||
require.Equal(t, 2, client.Entity.GetX(ctx, existing.ID).ReferenceCount)
|
||||
newFile := client.File.Query().Where(entfile.Name("copy.txt")).OnlyX(ctx)
|
||||
require.Equal(t, existing.ID, newFile.PrimaryEntity)
|
||||
require.Equal(t, int64(1024), newFile.Size)
|
||||
|
||||
// Size mismatch -> normal session
|
||||
s, err = f.PrepareUpload(ctx, uploadReq(t, f.hasher, u, "bigger.txt", 4096, dedupTestHash))
|
||||
require.NoError(t, err)
|
||||
require.False(t, s.Props.RapidUploaded)
|
||||
}
|
||||
|
||||
func TestPrepareUploadRapidScopeOff(t *testing.T) {
|
||||
client := enttest.Open(t, "sqlite3", "file:"+t.Name()+"?mode=memory&cache=shared")
|
||||
t.Cleanup(func() { require.NoError(t, client.Close()) })
|
||||
ctx := context.Background()
|
||||
|
||||
u, p, f := dedupUploadFixture(t, client, "off")
|
||||
seedCompletedEntity(t, client, u, p, dedupTestHash, 1024)
|
||||
|
||||
s, err := f.PrepareUpload(ctx, uploadReq(t, f.hasher, u, "copy.txt", 1024, dedupTestHash))
|
||||
require.NoError(t, err)
|
||||
require.False(t, s.Props.RapidUploaded)
|
||||
}
|
||||
Loading…
Reference in new issue