royalcat/tstor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

Compare commits

base: royalcat:bc4b39b1c123e7f8f9ddeb8e6829beb4c48c674b
Branches Tags
royalcat:master
..
compare: royalcat:e9df8925d1416d216670fbed55605aed9c12e745
Branches Tags
royalcat:master

No commits in common. "bc4b39b1c123e7f8f9ddeb8e6829beb4c48c674b" and "e9df8925d1416d216670fbed55605aed9c12e745" have entirely different histories.
bc4b39b1c1 ... e9df8925d1

41 changed files with 201 additions and 273 deletions
Show stats Expand all files Collapse all files

4
go.mod
View file

@ -35,8 +35,8 @@ require (
github.com/ravilushqa/otelgqlgen v0.15.0
github.com/royalcat/ctxio v0.0.0-20240602060200-590d464c39be
github.com/royalcat/ctxprogress v0.0.0-20240614113930-3cc5bb935bff
github.com/royalcat/kv v0.0.0-20240615090409-961d9afa99b6
github.com/royalcat/kv/kvbadger v0.0.0-20240615090409-961d9afa99b6
github.com/royalcat/kv v0.0.0-20240612224509-6aa0da315950
github.com/royalcat/kv/kvbadger v0.0.0-20240612224509-6aa0da315950
github.com/rs/zerolog v1.32.0
github.com/samber/slog-multi v1.0.2
github.com/samber/slog-zerolog v1.0.0

8
go.sum
View file

@ -541,14 +541,14 @@ github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU
github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
github.com/royalcat/ctxio v0.0.0-20240602060200-590d464c39be h1:Ui+Imq1Vk26rfpkLUsgvVdYO/UOJkzDyPzESfrTqWfM=
github.com/royalcat/ctxio v0.0.0-20240602060200-590d464c39be/go.mod h1:NFNp3OsEMUPYj5LZUFDiyDt+2E6gR/g8JLd0k+y8XWI=
github.com/royalcat/ctxprogress v0.0.0-20240511091748-6d9b327537c3 h1:1Ow/NUAWFZLghFcdNuyHt5Avb+bEI11qG8ELr9/XmQQ=
github.com/royalcat/ctxprogress v0.0.0-20240511091748-6d9b327537c3/go.mod h1:RcUpbosy/m3bJ3JsVO18MXEbrKRHOHkmYBXigDGekaA=
github.com/royalcat/ctxprogress v0.0.0-20240614113930-3cc5bb935bff h1:KlZaOEZYhCzyNYIp0LcE7MNR2Ar0PJS3eJU6A5mMTpk=
github.com/royalcat/ctxprogress v0.0.0-20240614113930-3cc5bb935bff/go.mod h1:RcUpbosy/m3bJ3JsVO18MXEbrKRHOHkmYBXigDGekaA=
github.com/royalcat/kv v0.0.0-20240615090409-961d9afa99b6 h1:rGXhPFpOVLeOO/Da2qxBNZY5yaQdTCGxQV2dUDXXf7U=
github.com/royalcat/kv v0.0.0-20240615090409-961d9afa99b6/go.mod h1:UB/VwpTut8c3IXLJFvYWFxAAZymk9eBuJRMJmpSpwYU=
github.com/royalcat/kv v0.0.0-20240612224509-6aa0da315950 h1:zHYwRhGWVkGQnjmStcnxTQ95Mtk5DL6w1PmdIn63EpI=
github.com/royalcat/kv v0.0.0-20240612224509-6aa0da315950/go.mod h1:UB/VwpTut8c3IXLJFvYWFxAAZymk9eBuJRMJmpSpwYU=
github.com/royalcat/kv/kvbadger v0.0.0-20240612224509-6aa0da315950 h1:rKG2P4TNLgA4/Jl7LPayifjcw4txVGVSPkpHVhn3wnw=
github.com/royalcat/kv/kvbadger v0.0.0-20240612224509-6aa0da315950/go.mod h1:JxgA1VGwbqu+WqdmjmjT0v6KeWoWlN6Y5lesjmphExM=
github.com/royalcat/kv/kvbadger v0.0.0-20240615090409-961d9afa99b6 h1:/TWa41uAL8Vk0MkvZc03EjA1/bS2otK5q0/+6bSWKJI=
github.com/royalcat/kv/kvbadger v0.0.0-20240615090409-961d9afa99b6/go.mod h1:JxgA1VGwbqu+WqdmjmjT0v6KeWoWlN6Y5lesjmphExM=
github.com/rs/dnscache v0.0.0-20230804202142-fc85eb664529 h1:18kd+8ZUlt/ARXhljq+14TwAoKa61q6dX8jtwOf6DH8=
github.com/rs/dnscache v0.0.0-20230804202142-fc85eb664529/go.mod h1:qe5TWALJ8/a1Lqznoc5BDHpYX/8HU60Hm2AwRmqzxqA=
github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=

10
pkg/go-nfs/conn.go
View file

@ -117,16 +117,9 @@ func (c *conn) serializeWrites(ctx context.Context) {
// Handle a request. errors from this method indicate a failure to read or
// write on the network stream, and trigger a disconnection of the connection.
func (c *conn) handle(ctx context.Context, w *response) (err error) {
func (c *conn) handle(ctx context.Context, w *response) error {
ctx, span := tracer.Start(ctx, fmt.Sprintf("nfs.handle.%s", NFSProcedure(w.req.Header.Proc).String()))
defer span.End()
defer func() {
if err != nil {
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
}()
handler := c.Server.handlerFor(w.req.Header.Prog, w.req.Header.Proc)
if handler == nil {
@ -153,6 +146,7 @@ func (c *conn) handle(ctx context.Context, w *response) (err error) {
}
}
span.SetStatus(codes.Ok, "")
return nil
}

6
pkg/go-nfs/handler.go
View file

@ -25,9 +25,9 @@ type Handler interface {
// represent file objects as opaque references
// Can be safely implemented via helpers/cachinghandler.
ToHandle(ctx context.Context, fs Filesystem, path []string) []byte
FromHandle(ctx context.Context, fh []byte) (Filesystem, []string, error)
InvalidateHandle(context.Context, Filesystem, []byte) error
ToHandle(fs Filesystem, path []string) []byte
FromHandle(fh []byte) (Filesystem, []string, error)
InvalidateHandle(Filesystem, []byte) error
// How many handles can be safely maintained by the handler.
HandleLimit() int

7
pkg/go-nfs/helpers/cachinghandler.go
View file

@ -1,7 +1,6 @@
package helpers
import (
"context"
"crypto/sha256"
"encoding/binary"
"io/fs"
@ -52,7 +51,7 @@ type entry struct {
// ToHandle takes a file and represents it with an opaque handle to reference it.
// In stateless nfs (when it's serving a unix fs) this can be the device + inode
// but we can generalize with a stateful local cache of handed out IDs.
func (c *CachingHandler) ToHandle(ctx context.Context, f nfs.Filesystem, path []string) []byte {
func (c *CachingHandler) ToHandle(f nfs.Filesystem, path []string) []byte {
joinedPath := f.Join(path...)
if handle := c.searchReverseCache(f, joinedPath); handle != nil {
@ -80,7 +79,7 @@ func (c *CachingHandler) ToHandle(ctx context.Context, f nfs.Filesystem, path []
}
// FromHandle converts from an opaque handle to the file it represents
func (c *CachingHandler) FromHandle(ctx context.Context, fh []byte) (nfs.Filesystem, []string, error) {
func (c *CachingHandler) FromHandle(fh []byte) (nfs.Filesystem, []string, error) {
id, err := uuid.FromBytes(fh)
if err != nil {
return nil, []string{}, err
@ -135,7 +134,7 @@ func (c *CachingHandler) evictReverseCache(path string, handle uuid.UUID) {
}
}
func (c *CachingHandler) InvalidateHandle(ctx context.Context, fs nfs.Filesystem, handle []byte) error {
func (c *CachingHandler) InvalidateHandle(fs nfs.Filesystem, handle []byte) error {
//Remove from cache
id, _ := uuid.FromBytes(handle)
entry, ok := c.activeHandles.Get(id)

8
pkg/go-nfs/helpers/nullauthhandler.go
View file

@ -35,21 +35,21 @@ func (h *NullAuthHandler) Change(fs nfs.Filesystem) nfs.Change {
}
// FSStat provides information about a filesystem.
func (h *NullAuthHandler) FSStat(context.Context, nfs.Filesystem, *nfs.FSStat) error {
func (h *NullAuthHandler) FSStat(ctx context.Context, f nfs.Filesystem, s *nfs.FSStat) error {
return nil
}
// ToHandle handled by CachingHandler
func (h *NullAuthHandler) ToHandle(context.Context, nfs.Filesystem, []string) []byte {
func (h *NullAuthHandler) ToHandle(f nfs.Filesystem, s []string) []byte {
return []byte{}
}
// FromHandle handled by CachingHandler
func (h *NullAuthHandler) FromHandle(context.Context, []byte) (nfs.Filesystem, []string, error) {
func (h *NullAuthHandler) FromHandle([]byte) (nfs.Filesystem, []string, error) {
return nil, []string{}, nil
}
func (c *NullAuthHandler) InvalidateHandle(context.Context, nfs.Filesystem, []byte) error {
func (c *NullAuthHandler) InvalidateHandle(nfs.Filesystem, []byte) error {
return nil
}

2
pkg/go-nfs/mount.go
View file

@ -39,7 +39,7 @@ func onMount(ctx context.Context, w *response, userHandle Handler) error {
return err
}
rootHndl := userHandle.ToHandle(ctx, handle, []string{})
rootHndl := userHandle.ToHandle(handle, []string{})
if status == MountStatusOk {
_ = xdr.Write(writer, rootHndl)

2
pkg/go-nfs/nfs_onaccess.go
View file

@ -14,7 +14,7 @@ func onAccess(ctx context.Context, w *response, userHandle Handler) error {
if err != nil {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, roothandle)
fs, path, err := userHandle.FromHandle(roothandle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}

2
pkg/go-nfs/nfs_oncommit.go
View file

@ -18,7 +18,7 @@ func onCommit(ctx context.Context, w *response, userHandle Handler) error {
}
// The conn will drain the unread offset and count arguments.
fs, path, err := userHandle.FromHandle(ctx, handle)
fs, path, err := userHandle.FromHandle(handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}

4
pkg/go-nfs/nfs_oncreate.go
View file

@ -47,7 +47,7 @@ func onCreate(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusNotSupp, os.ErrInvalid}
}
fs, path, err := userHandle.FromHandle(ctx, obj.Handle)
fs, path, err := userHandle.FromHandle(obj.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -87,7 +87,7 @@ func onCreate(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusAccess, err}
}
fp := userHandle.ToHandle(ctx, fs, newFile)
fp := userHandle.ToHandle(fs, newFile)
changer := userHandle.Change(fs)
if err := attrs.Apply(ctx, changer, fs, newFilePath); err != nil {
Log.Errorf("Error applying attributes: %v\n", err)

2
pkg/go-nfs/nfs_onfsinfo.go
View file

@ -24,7 +24,7 @@ func onFSInfo(ctx context.Context, w *response, userHandle Handler) error {
if err != nil {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, roothandle)
fs, path, err := userHandle.FromHandle(roothandle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}

2
pkg/go-nfs/nfs_onfsstat.go
View file

@ -13,7 +13,7 @@ func onFSStat(ctx context.Context, w *response, userHandle Handler) error {
if err != nil {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, roothandle)
fs, path, err := userHandle.FromHandle(roothandle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}

2
pkg/go-nfs/nfs_ongetattr.go
View file

@ -15,7 +15,7 @@ func onGetAttr(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, handle)
fs, path, err := userHandle.FromHandle(handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}

4
pkg/go-nfs/nfs_onlink.go
View file

@ -27,7 +27,7 @@ func onLink(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, obj.Handle)
fs, path, err := userHandle.FromHandle(obj.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -49,7 +49,7 @@ func onLink(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusNotDir, nil}
}
fp := userHandle.ToHandle(ctx, fs, append(path, string(obj.Filename)))
fp := userHandle.ToHandle(fs, append(path, string(obj.Filename)))
changer := userHandle.Change(fs)
if changer == nil {
return &NFSStatusError{NFSStatusAccess, err}

6
pkg/go-nfs/nfs_onlookup.go
View file

@ -33,7 +33,7 @@ func onLookup(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusInval, err}
}
fs, p, err := userHandle.FromHandle(ctx, obj.Handle)
fs, p, err := userHandle.FromHandle(obj.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -58,7 +58,7 @@ func onLookup(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusAccess, os.ErrPermission}
}
pPath := p[0 : len(p)-1]
pHandle := userHandle.ToHandle(ctx, fs, pPath)
pHandle := userHandle.ToHandle(fs, pPath)
resp, err := lookupSuccessResponse(ctx, pHandle, pPath, p, fs)
if err != nil {
return &NFSStatusError{NFSStatusServerFault, err}
@ -74,7 +74,7 @@ func onLookup(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusNoEnt, os.ErrNotExist}
}
newHandle := userHandle.ToHandle(ctx, fs, reqPath)
newHandle := userHandle.ToHandle(fs, reqPath)
resp, err := lookupSuccessResponse(ctx, newHandle, reqPath, p, fs)
if err != nil {
return &NFSStatusError{NFSStatusServerFault, err}

4
pkg/go-nfs/nfs_onmkdir.go
View file

@ -26,7 +26,7 @@ func onMkdir(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, obj.Handle)
fs, path, err := userHandle.FromHandle(obj.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -59,7 +59,7 @@ func onMkdir(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusAccess, err}
}
fp := userHandle.ToHandle(ctx, fs, newFolder)
fp := userHandle.ToHandle(fs, newFolder)
changer := userHandle.Change(fs)
if changer != nil {
if err := attrs.Apply(ctx, changer, fs, newFolderPath); err != nil {

4
pkg/go-nfs/nfs_onmknod.go
View file

@ -37,7 +37,7 @@ func onMknod(ctx context.Context, w *response, userHandle Handler) error {
}
// see if the filesystem supports mknod
fs, path, err := userHandle.FromHandle(ctx, obj.Handle)
fs, path, err := userHandle.FromHandle(obj.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -69,7 +69,7 @@ func onMknod(ctx context.Context, w *response, userHandle Handler) error {
} else if !parent.IsDir() {
return &NFSStatusError{NFSStatusNotDir, nil}
}
fp := userHandle.ToHandle(ctx, fs, append(path, string(obj.Filename)))
fp := userHandle.ToHandle(fs, append(path, string(obj.Filename)))
switch nfs_ftype(ftype) {
case FTYPE_NF3CHR:

2
pkg/go-nfs/nfs_onpathconf.go
View file

@ -15,7 +15,7 @@ func onPathConf(ctx context.Context, w *response, userHandle Handler) error {
if err != nil {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, roothandle)
fs, path, err := userHandle.FromHandle(roothandle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}

2
pkg/go-nfs/nfs_onread.go
View file

@ -37,7 +37,7 @@ func onRead(ctx context.Context, w *response, userHandle Handler) error {
if err != nil {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, obj.Handle)
fs, path, err := userHandle.FromHandle(obj.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}

4
pkg/go-nfs/nfs_onreaddir.go
View file

@ -41,7 +41,7 @@ func onReadDir(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusTooSmall, io.ErrShortBuffer}
}
fs, p, err := userHandle.FromHandle(ctx, obj.Handle)
fs, p, err := userHandle.FromHandle(obj.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -140,7 +140,7 @@ func onReadDir(ctx context.Context, w *response, userHandle Handler) error {
func getDirListingWithVerifier(ctx context.Context, userHandle Handler, fsHandle []byte, verifier uint64) ([]fs.FileInfo, uint64, error) {
// figure out what directory it is.
fs, p, err := userHandle.FromHandle(ctx, fsHandle)
fs, p, err := userHandle.FromHandle(fsHandle)
if err != nil {
return nil, 0, &NFSStatusError{NFSStatusStale, err}
}

4
pkg/go-nfs/nfs_onreaddirplus.go
View file

@ -46,7 +46,7 @@ func onReadDirPlus(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusTooSmall, nil}
}
fs, p, err := userHandle.FromHandle(ctx, obj.Handle)
fs, p, err := userHandle.FromHandle(obj.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -102,7 +102,7 @@ func onReadDirPlus(ctx context.Context, w *response, userHandle Handler) error {
}
filePath := joinPath(p, c.Name())
handle := userHandle.ToHandle(ctx, fs, filePath)
handle := userHandle.ToHandle(fs, filePath)
attrs := ToFileAttribute(c, path.Join(filePath...))
entities = append(entities, readDirPlusEntity{
FileID: attrs.Fileid,

2
pkg/go-nfs/nfs_onreadlink.go
View file

@ -15,7 +15,7 @@ func onReadLink(ctx context.Context, w *response, userHandle Handler) error {
if err != nil {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, handle)
fs, path, err := userHandle.FromHandle(handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}

6
pkg/go-nfs/nfs_onremove.go
View file

@ -15,7 +15,7 @@ func onRemove(ctx context.Context, w *response, userHandle Handler) error {
if err := xdr.Read(w.req.Body, &obj); err != nil {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, obj.Handle)
fs, path, err := userHandle.FromHandle(obj.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -49,7 +49,7 @@ func onRemove(ctx context.Context, w *response, userHandle Handler) error {
preCacheData := ToFileAttribute(dirInfo, fullPath).AsCache()
toDelete := fs.Join(append(path, string(obj.Filename))...)
toDeleteHandle := userHandle.ToHandle(ctx, fs, append(path, string(obj.Filename)))
toDeleteHandle := userHandle.ToHandle(fs, append(path, string(obj.Filename)))
err = fs.Remove(ctx, toDelete)
if err != nil {
@ -65,7 +65,7 @@ func onRemove(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusIO, err}
}
if err := userHandle.InvalidateHandle(ctx, fs, toDeleteHandle); err != nil {
if err := userHandle.InvalidateHandle(fs, toDeleteHandle); err != nil {
return &NFSStatusError{NFSStatusServerFault, err}
}

8
pkg/go-nfs/nfs_onrename.go
View file

@ -20,7 +20,7 @@ func onRename(ctx context.Context, w *response, userHandle Handler) error {
if err != nil {
return &NFSStatusError{NFSStatusInval, err}
}
fs, fromPath, err := userHandle.FromHandle(ctx, from.Handle)
fs, fromPath, err := userHandle.FromHandle(from.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -29,7 +29,7 @@ func onRename(ctx context.Context, w *response, userHandle Handler) error {
if err = xdr.Read(w.req.Body, &to); err != nil {
return &NFSStatusError{NFSStatusInval, err}
}
fs2, toPath, err := userHandle.FromHandle(ctx, to.Handle)
fs2, toPath, err := userHandle.FromHandle(to.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -78,7 +78,7 @@ func onRename(ctx context.Context, w *response, userHandle Handler) error {
}
preDestData := ToFileAttribute(toDirInfo, toDirPath).AsCache()
oldHandle := userHandle.ToHandle(ctx, fs, append(fromPath, string(from.Filename)))
oldHandle := userHandle.ToHandle(fs, append(fromPath, string(from.Filename)))
fromLoc := fs.Join(append(fromPath, string(from.Filename))...)
toLoc := fs.Join(append(toPath, string(to.Filename))...)
@ -97,7 +97,7 @@ func onRename(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusIO, err}
}
if err := userHandle.InvalidateHandle(ctx, fs, oldHandle); err != nil {
if err := userHandle.InvalidateHandle(fs, oldHandle); err != nil {
return &NFSStatusError{NFSStatusServerFault, err}
}

2
pkg/go-nfs/nfs_onsetattr.go
View file

@ -17,7 +17,7 @@ func onSetAttr(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, handle)
fs, path, err := userHandle.FromHandle(handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}

4
pkg/go-nfs/nfs_onsymlink.go
View file

@ -26,7 +26,7 @@ func onSymlink(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, obj.Handle)
fs, path, err := userHandle.FromHandle(obj.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}
@ -53,7 +53,7 @@ func onSymlink(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusAccess, err}
}
fp := userHandle.ToHandle(ctx, fs, append(path, string(obj.Filename)))
fp := userHandle.ToHandle(fs, append(path, string(obj.Filename)))
changer := userHandle.Change(fs)
if changer != nil {
if err := attrs.Apply(ctx, changer, fs, newFilePath); err != nil {

2
pkg/go-nfs/nfs_onwrite.go
View file

@ -35,7 +35,7 @@ func onWrite(ctx context.Context, w *response, userHandle Handler) error {
return &NFSStatusError{NFSStatusInval, err}
}
fs, path, err := userHandle.FromHandle(ctx, req.Handle)
fs, path, err := userHandle.FromHandle(req.Handle)
if err != nil {
return &NFSStatusError{NFSStatusStale, err}
}

2
pkg/kvsingle/single.go
View file

@ -15,7 +15,7 @@ func New[K, V any](db kv.Store[K, V], key K) *Value[K, V] {
return &Value[K, V]{Key: key, db: db}
}
func (s *Value[K, V]) Get(ctx context.Context) (V, error) {
func (s *Value[K, V]) Get(ctx context.Context) (V, bool, error) {
return s.db.Get(ctx, s.Key)
}

90
pkg/kvtrace/kvmetrics.go
View file

@ -6,145 +6,85 @@ import (
"github.com/royalcat/kv"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
)
var tracer = otel.Tracer("github.com/royalcat/kv/tracer")
type traceStore[K, V any] struct {
type traceSrtore[K, V any] struct {
kv kv.Store[K, V]
attrs []attribute.KeyValue
}
func WrapTracing[K, V any](kv kv.Store[K, V], attrs ...attribute.KeyValue) kv.Store[K, V] {
return &traceStore[K, V]{
return &traceSrtore[K, V]{
kv: kv,
attrs: attrs,
}
}
// Close implements kv.Store.
func (m *traceStore[K, V]) Close(ctx context.Context) (err error) {
func (m *traceSrtore[K, V]) Close(ctx context.Context) error {
ctx, span := tracer.Start(ctx, "Close", trace.WithAttributes(m.attrs...))
defer span.End()
defer func() {
if err != nil && err != kv.ErrKeyNotFound {
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
}()
return m.kv.Close(ctx)
}
// Delete implements kv.Store.
func (m *traceStore[K, V]) Delete(ctx context.Context, k K) (err error) {
func (m *traceSrtore[K, V]) Delete(ctx context.Context, k K) error {
ctx, span := tracer.Start(ctx, "Delete", trace.WithAttributes(m.attrs...))
defer span.End()
defer func() {
if err != nil && err != kv.ErrKeyNotFound {
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
}()
return m.kv.Delete(ctx, k)
}
// Get implements kv.Store.
func (m *traceStore[K, V]) Get(ctx context.Context, k K) (v V, err error) {
func (m *traceSrtore[K, V]) Get(ctx context.Context, k K) (v V, found bool, err error) {
ctx, span := tracer.Start(ctx, "Get", trace.WithAttributes(m.attrs...))
defer span.End()
defer func() {
if err != nil && err != kv.ErrKeyNotFound {
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
}()
return m.kv.Get(ctx, k)
}
// Get implements kv.Store.
func (m *traceStore[K, V]) Edit(ctx context.Context, k K, edit kv.Edit[V]) (err error) {
ctx, span := tracer.Start(ctx, "Get", trace.WithAttributes(m.attrs...))
defer span.End()
defer func() {
if err != nil && err != kv.ErrKeyNotFound {
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
}()
return m.kv.Edit(ctx, k, edit)
}
// Range implements kv.Store.
func (m *traceStore[K, V]) Range(ctx context.Context, iter kv.Iter[K, V]) (err error) {
func (m *traceSrtore[K, V]) Range(ctx context.Context, iter kv.Iter[K, V]) error {
ctx, span := tracer.Start(ctx, "Range", trace.WithAttributes(m.attrs...))
defer span.End()
defer func() {
if err != nil {
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
}()
count := 0
iterCount := func(k K, v V) error {
count++
return iter(k, v)
}
defer func() {
span.SetAttributes(attribute.Int("count", count))
}()
return m.kv.Range(ctx, iterCount)
err := m.kv.Range(ctx, iterCount)
span.SetAttributes(attribute.Int("count", count))
return err
}
// RangeWithPrefix implements kv.Store.
func (m *traceStore[K, V]) RangeWithPrefix(ctx context.Context, k K, iter kv.Iter[K, V]) (err error) {
func (m *traceSrtore[K, V]) RangeWithPrefix(ctx context.Context, k K, iter kv.Iter[K, V]) error {
ctx, span := tracer.Start(ctx, "RangeWithPrefix", trace.WithAttributes(m.attrs...))
defer span.End()
defer func() {
if err != nil {
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
}()
count := 0
iterCount := func(k K, v V) error {
count++
return iter(k, v)
}
defer func() {
span.SetAttributes(attribute.Int("count", count))
}()
return m.kv.Range(ctx, iterCount)
err := m.kv.Range(ctx, iterCount)
span.SetAttributes(attribute.Int("count", count))
return err
}
// Set implements kv.Store.
func (m *traceStore[K, V]) Set(ctx context.Context, k K, v V) (err error) {
func (m *traceSrtore[K, V]) Set(ctx context.Context, k K, v V) error {
ctx, span := tracer.Start(ctx, "Set", trace.WithAttributes(m.attrs...))
defer span.End()
defer func() {
if err != nil {
span.SetStatus(codes.Error, err.Error())
} else {
span.SetStatus(codes.Ok, "")
}
}()
return m.kv.Set(ctx, k, v)
}
var _ kv.Store[any, any] = (*traceStore[any, any])(nil)
var _ kv.Store[any, any] = (*traceSrtore[any, any])(nil)

24
src/export/nfs/kvhandler.go
View file

@ -2,7 +2,6 @@ package nfs
import (
"context"
"errors"
"fmt"
"path"
"strings"
@ -11,7 +10,6 @@ import (
"git.kmsign.ru/royalcat/tstor/pkg/go-nfs"
"git.kmsign.ru/royalcat/tstor/src/config"
"git.kmsign.ru/royalcat/tstor/src/log"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/metric"
@ -52,7 +50,6 @@ var kvhandlerMeter = otel.Meter("git.kmsign.ru/royalcat/tstor/src/export/nfs.kvh
func NewKvHandler(h nfs.Handler, fs nfs.Filesystem, config config.NFS) (nfs.Handler, error) {
opts := kvbadger.DefaultOptions(path.Join(config.CachePath, "handlers"))
opts.DefaultTTL = time.Hour
opts.BadgerOptions.Logger = log.BadgerLogger("nfs", "kvhandler")
activeHandles, err := kvbadger.NewBagerKVBinaryKey[uuid.UUID, handle](opts)
if err != nil {
@ -101,7 +98,8 @@ type CachingHandler struct {
// ToHandle takes a file and represents it with an opaque handle to reference it.
// In stateless nfs (when it's serving a unix fs) this can be the device + inode
// but we can generalize with a stateful local cache of handed out IDs.
func (c *CachingHandler) ToHandle(ctx context.Context, _ nfs.Filesystem, path []string) []byte {
func (c *CachingHandler) ToHandle(_ nfs.Filesystem, path []string) []byte {
ctx := context.Background()
var id uuid.UUID
cacheKey := handle(path).String()
@ -125,29 +123,31 @@ func (c *CachingHandler) ToHandle(ctx context.Context, _ nfs.Filesystem, path []
}
// FromHandle converts from an opaque handle to the file it represents
func (c *CachingHandler) FromHandle(ctx context.Context, fh []byte) (nfs.Filesystem, []string, error) {
func (c *CachingHandler) FromHandle(fh []byte) (nfs.Filesystem, []string, error) {
c.mu.Lock()
defer c.mu.Unlock()
ctx := context.Background()
id, err := uuid.FromBytes(fh)
if err != nil {
return nil, nil, err
}
paths, err := c.activeHandles.Get(ctx, id)
paths, found, err := c.activeHandles.Get(ctx, id)
if err != nil {
if errors.Is(err, kv.ErrKeyNotFound) {
return nil, nil, &nfs.NFSStatusError{NFSStatus: nfs.NFSStatusStale}
}
return nil, nil, fmt.Errorf("kv error: %w", err)
}
if found {
return c.fs, paths, nil
}
func (c *CachingHandler) InvalidateHandle(ctx context.Context, fs nfs.Filesystem, handle []byte) error {
return nil, nil, &nfs.NFSStatusError{NFSStatus: nfs.NFSStatusStale}
}
func (c *CachingHandler) InvalidateHandle(fs nfs.Filesystem, handle []byte) error {
ctx := context.Background()
//Remove from cache
id, err := uuid.FromBytes(handle)
if err != nil {

19
src/log/badger.go
View file

@ -5,19 +5,12 @@ import (
"log/slog"
"strings"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"github.com/dgraph-io/badger/v4"
)
func BadgerLogger(name ...string) badger.Logger {
return &badgerLogger{
L: rlog.Component(append(name, "badger")...).Slog(),
}
}
var _ badger.Logger = (*Badger)(nil)
var _ badger.Logger = (*badgerLogger)(nil)
type badgerLogger struct {
type Badger struct {
L *slog.Logger
}
@ -25,18 +18,18 @@ func fmtBadgerLog(m string, f ...any) string {
return fmt.Sprintf(strings.ReplaceAll(m, "\n", ""), f...)
}
func (l *badgerLogger) Errorf(m string, f ...interface{}) {
func (l *Badger) Errorf(m string, f ...interface{}) {
l.L.Error(fmtBadgerLog(m, f...))
}
func (l *badgerLogger) Warningf(m string, f ...interface{}) {
func (l *Badger) Warningf(m string, f ...interface{}) {
l.L.Warn(fmtBadgerLog(m, f...))
}
func (l *badgerLogger) Infof(m string, f ...interface{}) {
func (l *Badger) Infof(m string, f ...interface{}) {
l.L.Info(fmtBadgerLog(m, f...))
}
func (l *badgerLogger) Debugf(m string, f ...interface{}) {
func (l *Badger) Debugf(m string, f ...interface{}) {
l.L.Debug(fmtBadgerLog(m, f...))
}

83
src/sources/torrent/controller.go
View file

@ -8,33 +8,20 @@ import (
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/types"
"github.com/royalcat/kv"
)
type TorrentFileDeleter interface {
DeleteFile(file *torrent.File) error
}
type FileProperties struct {
Excluded bool `json:"excluded"`
Priority types.PiecePriority `json:"priority"`
}
type Controller struct {
torrentFilePath string
t *torrent.Torrent
storage TorrentFileDeleter
fileProperties kv.Store[string, FileProperties]
rep *filesMappingsStore
log *rlog.Logger
}
func newController(t *torrent.Torrent, fileProperties kv.Store[string, FileProperties], storage TorrentFileDeleter) *Controller {
func newController(t *torrent.Torrent, rep *filesMappingsStore) *Controller {
return &Controller{
t: t,
storage: storage,
fileProperties: fileProperties,
log: rlog.Component("torrent-client", "controller").With(slog.String("infohash", t.InfoHash().HexString())),
rep: rep,
log: rlog.Component("torrent/controller").With(slog.String("infohash", t.InfoHash().HexString())),
}
}
@ -76,11 +63,7 @@ func (s *Controller) Length() int64 {
}
func (s *Controller) Files(ctx context.Context) ([]*torrent.File, error) {
fps := map[string]FileProperties{}
err := s.fileProperties.Range(ctx, func(k string, v FileProperties) error {
fps[k] = v
return nil
})
fileMappings, err := s.rep.FileMappings(ctx, s.t.InfoHash())
if err != nil {
return nil, err
}
@ -101,10 +84,9 @@ func (s *Controller) Files(ctx context.Context) ([]*torrent.File, error) {
if strings.Contains(p, "/.pad/") {
return true
}
if props, ok := fps[p]; ok && props.Excluded {
if target, ok := fileMappings[p]; ok && target == "" {
return true
}
return false
})
@ -120,13 +102,7 @@ func Map[T, U any](ts []T, f func(T) U) []U {
}
func (s *Controller) ExcludeFile(ctx context.Context, f *torrent.File) error {
log := s.log.With(slog.String("file", f.Path()))
log.Info(ctx, "excluding file")
return s.fileProperties.Edit(ctx, f.Path(), func(ctx context.Context, v FileProperties) (FileProperties, error) {
v.Excluded = true
return v, nil
})
return s.rep.ExcludeFile(ctx, f)
}
func (s *Controller) isFileComplete(startIndex int, endIndex int) bool {
@ -156,46 +132,21 @@ func (s *Controller) ValidateTorrent(ctx context.Context) error {
return nil
}
func (c *Controller) SetFilePriority(ctx context.Context, file *torrent.File, priority types.PiecePriority) error {
log := c.log.With(slog.String("file", file.Path()), slog.Int("priority", int(priority)))
log.Info(ctx, "set pritority for file")
err := c.fileProperties.Edit(ctx, file.Path(), func(ctx context.Context, v FileProperties) (FileProperties, error) {
v.Priority = priority
return v, nil
})
if err != nil {
return err
}
file.SetPriority(priority)
return nil
}
func (c *Controller) initializeTorrentPriories(ctx context.Context) error {
log := c.log.WithComponent("initializeTorrentPriories")
files, err := c.Files(ctx)
if err != nil {
return err
}
// files, err := c.Files(ctx)
// if err != nil {
// return err
// }
for _, file := range files {
if file == nil {
continue
}
// for _, file := range files {
// if file == nil {
// continue
// }
props, err := c.fileProperties.Get(ctx, file.Path())
if err != nil {
if err == kv.ErrKeyNotFound {
continue
}
log.Error(ctx, "failed to get file priority", rlog.Error(err))
}
file.SetPriority(props.Priority)
}
// file.SetPriority(torrent.PiecePriorityNormal)
// }
log.Info(ctx, "torrent initialization complete", slog.String("infohash", c.InfoHash()), slog.String("torrent_name", c.Name()))

27
src/sources/torrent/daemon.go
View file

@ -15,7 +15,6 @@ import (
"git.kmsign.ru/royalcat/tstor/pkg/ctxbilly"
"git.kmsign.ru/royalcat/tstor/pkg/rlog"
"git.kmsign.ru/royalcat/tstor/src/config"
"git.kmsign.ru/royalcat/tstor/src/tkv"
"git.kmsign.ru/royalcat/tstor/src/vfs"
"github.com/royalcat/ctxio"
"go.opentelemetry.io/otel"
@ -41,11 +40,11 @@ type DirAquire struct {
type Daemon struct {
client *torrent.Client
excludedFiles *filesMappingsStore
infoBytes *infoBytesStore
Storage *fileStorage
fis *fileItemStore
dirsAquire kv.Store[string, DirAquire]
fileProperties kv.Store[string, FileProperties]
loadMutex sync.Mutex
torrentLoaded chan struct{}
@ -78,13 +77,12 @@ func NewService(sourceFs billy.Filesystem, conf config.TorrentClient) (*Daemon,
return nil, err
}
s.fileProperties, err = tkv.NewKV[string, FileProperties](conf.MetadataFolder, "file-properties")
s.excludedFiles, err = newFileMappingsStore(conf.MetadataFolder, s.Storage)
if err != nil {
return nil, err
}
s.infoBytes, err = newInfoBytesStore(conf.MetadataFolder)
if err != nil {
return nil, err
}
@ -100,7 +98,7 @@ func NewService(sourceFs billy.Filesystem, conf config.TorrentClient) (*Daemon,
}
s.client.AddDhtNodes(conf.DHTNodes)
s.dirsAquire, err = tkv.NewKV[string, DirAquire](conf.MetadataFolder, "dir-acquire")
s.dirsAquire, err = NewKV[string, DirAquire](conf.MetadataFolder, "dir-acquire")
if err != nil {
return nil, err
}
@ -124,7 +122,7 @@ func (s *Daemon) Close(ctx context.Context) error {
s.client.Close(),
s.Storage.Close(),
s.dirsAquire.Close(ctx),
// s.excludedFiles.Close(ctx),
s.excludedFiles.Close(ctx),
s.infoBytes.Close(),
s.fis.Close(),
)...)
@ -216,7 +214,7 @@ func (s *Daemon) LoadTorrent(ctx context.Context, f vfs.File) (*Controller, erro
// }
}
ctl := s.newController(t)
ctl := newController(t, s.excludedFiles)
err = ctl.initializeTorrentPriories(ctx)
if err != nil {
@ -308,23 +306,12 @@ func (s *Daemon) loadTorrentFiles(ctx context.Context) error {
})
}
func storeByTorrent[K kv.Bytes, V any](s kv.Store[K, V], infohash infohash.T) kv.Store[K, V] {
return kv.PrefixBytes[K, V](s, K(infohash.HexString()+"/"))
}
func (s *Daemon) newController(t *torrent.Torrent) *Controller {
return newController(t,
storeByTorrent(s.fileProperties, t.InfoHash()),
s.Storage,
)
}
func (s *Daemon) ListTorrents(ctx context.Context) ([]*Controller, error) {
<-s.torrentLoaded
out := []*Controller{}
for _, v := range s.client.Torrents() {
out = append(out, s.newController(v))
out = append(out, newController(v, s.excludedFiles))
}
return out, nil
}
@ -337,7 +324,7 @@ func (s *Daemon) GetTorrent(infohashHex string) (*Controller, error) {
return nil, nil
}
return s.newController(t), nil
return newController(t, s.excludedFiles), nil
}
func slicesUnique[S ~[]E, E comparable](in S) S {

60
src/sources/torrent/file_mappings.go Normal file
View file

@ -0,0 +1,60 @@
package torrent
import (
"context"
"path/filepath"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/types/infohash"
"github.com/royalcat/kv"
"github.com/royalcat/kv/kvbadger"
)
func newFileMappingsStore(metaDir string, storage TorrentFileDeleter) (*filesMappingsStore, error) {
opts := kvbadger.DefaultOptions(filepath.Join(metaDir, "file-mappings"))
str, err := kvbadger.NewBadgerKVBytes[string, string](opts)
if err != nil {
return nil, err
}
r := &filesMappingsStore{
mappings: str,
storage: storage,
}
return r, nil
}
type filesMappingsStore struct {
mappings kv.Store[string, string]
storage TorrentFileDeleter
}
type TorrentFileDeleter interface {
DeleteFile(file *torrent.File) error
}
func fileKey(file *torrent.File) string {
return file.Torrent().InfoHash().HexString() + "/" + file.Path()
}
func (r *filesMappingsStore) MapFile(ctx context.Context, file *torrent.File, target string) error {
return r.mappings.Set(ctx, fileKey(file), target)
}
func (r *filesMappingsStore) ExcludeFile(ctx context.Context, file *torrent.File) error {
return r.mappings.Set(ctx, fileKey(file), "")
}
func (r *filesMappingsStore) FileMappings(ctx context.Context, ih infohash.T) (map[string]string, error) {
out := map[string]string{}
err := r.mappings.RangeWithPrefix(ctx, ih.HexString(), func(k, v string) error {
out[k] = v
return nil
})
return out, err
}
func (r *filesMappingsStore) Close(ctx context.Context) error {
return r.mappings.Close(ctx)
}

5
src/sources/torrent/fileitem.go
View file

@ -3,6 +3,7 @@ package torrent
import (
"bytes"
"encoding/gob"
"log/slog"
"time"
dlog "git.kmsign.ru/royalcat/tstor/src/log"
@ -18,8 +19,10 @@ type fileItemStore struct {
}
func newFileItemStore(path string, itemsTTL time.Duration) (*fileItemStore, error) {
l := slog.With("component", "item-store")
opts := badger.DefaultOptions(path).
WithLogger(dlog.BadgerLogger("torrent-client", "item-store")).
WithLogger(&dlog.Badger{L: l}).
WithValueLogFileSize(1<<26 - 1)
db, err := badger.Open(opts)

5
src/sources/torrent/infobytes.go
View file

@ -4,6 +4,7 @@ import (
"bytes"
"errors"
"fmt"
"log/slog"
"path/filepath"
dlog "git.kmsign.ru/royalcat/tstor/src/log"
@ -19,9 +20,11 @@ type infoBytesStore struct {
}
func newInfoBytesStore(metaDir string) (*infoBytesStore, error) {
l := slog.With("component", "badger", "db", "info-bytes")
opts := badger.
DefaultOptions(filepath.Join(metaDir, "infobytes")).
WithLogger(dlog.BadgerLogger("torrent-client", "infobytes"))
WithLogger(&dlog.Badger{L: l})
db, err := badger.Open(opts)
if err != nil {
return nil, err

4
src/tkv/new.go → src/sources/torrent/kv.go
View file

@ -1,10 +1,9 @@
package tkv
package torrent
import (
"path"
"git.kmsign.ru/royalcat/tstor/pkg/kvtrace"
tlog "git.kmsign.ru/royalcat/tstor/src/log"
"github.com/royalcat/kv"
"github.com/royalcat/kv/kvbadger"
"go.opentelemetry.io/otel/attribute"
@ -12,7 +11,6 @@ import (
func NewKV[K kv.Bytes, V any](dbdir, name string) (store kv.Store[K, V], err error) {
opts := kvbadger.DefaultOptions(path.Join(dbdir, name))
opts.BadgerOptions.Logger = tlog.BadgerLogger(name, "badger")
store, err = kvbadger.NewBadgerKVBytesKey[K, V](opts)
if err != nil {
return nil, err

4
src/sources/torrent/piece_completion.go
View file

@ -33,9 +33,11 @@ type badgerPieceCompletion struct {
var _ storage.PieceCompletion = (*badgerPieceCompletion)(nil)
func newPieceCompletion(dir string) (storage.PieceCompletion, error) {
l := slog.With("component", "badger", "db", "piece-completion")
opts := badger.
DefaultOptions(dir).
WithLogger(dlog.BadgerLogger("torrent-client", "piece-completion"))
WithLogger(&dlog.Badger{L: l})
db, err := badger.Open(opts)
if err != nil {
return nil, err

4
src/sources/torrent/stats_store.go
View file

@ -6,7 +6,6 @@ import (
"path"
"time"
"git.kmsign.ru/royalcat/tstor/src/log"
"github.com/anacrolix/torrent/types/infohash"
"github.com/dgraph-io/badger/v4"
"golang.org/x/exp/maps"
@ -16,8 +15,7 @@ func newStatsStore(metaDir string, lifetime time.Duration) (*statsStore, error)
db, err := badger.OpenManaged(
badger.
DefaultOptions(path.Join(metaDir, "stats-history")).
WithNumVersionsToKeep(int(^uint(0) >> 1)).
WithLogger(log.BadgerLogger("stats")), // Infinity
WithNumVersionsToKeep(int(^uint(0) >> 1)), // Infinity
)
if err != nil {
return nil, err

11
src/sources/torrent/storage_open.go
View file

@ -16,7 +16,6 @@ import (
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/storage"
"github.com/anacrolix/torrent/types/infohash"
"github.com/royalcat/kv"
)
// OpenTorrent implements storage.ClientImplCloser.
@ -93,8 +92,11 @@ func (s *Daemon) checkTorrentCompatable(ctx context.Context, ih infohash.T, info
name := info.BestName()
aq, err := s.dirsAquire.Get(ctx, info.BestName())
if errors.Is(err, kv.ErrKeyNotFound) {
aq, found, err := s.dirsAquire.Get(ctx, info.BestName())
if err != nil {
return false, false, err
}
if !found {
err = s.dirsAquire.Set(ctx, name, DirAquire{
Name: name,
Hashes: slices.Compact([]infohash.T{ih}),
@ -105,8 +107,6 @@ func (s *Daemon) checkTorrentCompatable(ctx context.Context, ih infohash.T, info
log.Debug(ctx, "acquiring was not found, so created")
return true, false, nil
} else if err != nil {
return false, false, err
}
if slices.Contains(aq.Hashes, ih) {
@ -136,6 +136,7 @@ func (s *Daemon) checkTorrentCompatable(ctx context.Context, ih infohash.T, info
}
}
if slices.Contains(aq.Hashes, ih) {
log.Debug(ctx, "hash is compatable")
return true, false, nil

11
src/sources/ytdlp/controller.go
View file

@ -11,7 +11,6 @@ import (
"git.kmsign.ru/royalcat/tstor/src/tasks"
"github.com/royalcat/ctxio"
"github.com/royalcat/ctxprogress"
"github.com/royalcat/kv"
)
type Controller struct {
@ -87,13 +86,13 @@ func (c *Controller) Update(ctx context.Context, updater tasks.Updater) error {
}
func (c *Controller) Info(ctx context.Context) (ytdlp.Info, error) {
info, err := c.cachedinfo.Get(ctx)
if err == nil {
return info, nil
}
if err != kv.ErrKeyNotFound {
info, found, err := c.cachedinfo.Get(ctx)
if err != nil {
return info, err
}
if found {
return info, nil
}
info, err = c.Info(ctx)
if err != nil {