mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-28 03:46:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

tarantool: fix upsert data corruption and missing context propagation (#10926)

Browse Source 
Co-authored-by: Marat Karimov <karimov_m@inbox.ru>
This commit is contained in:
MaratKarimov
2026-08-24 15:17:08 -07:00
committed by GitHub
co-authored by Marat Karimov
parent a0ddf22f17
commit a3afe4460b
2 changed files with 18 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/filer/tarantool/tarantool_store.go
+11 -6
View File
@@ -143,7 +143,7 @@ func (store *TarantoolStore) InsertEntry(ctx context.Context, entry *filer.Entry
var operations = []crud.Operation{
{
Operator: crud.Insert,
Operator: crud.Assign,
Field: "data",
Value: string(meta),
},
@@ -151,7 +151,8 @@ func (store *TarantoolStore) InsertEntry(ctx context.Context, entry *filer.Entry
req := crud.NewUpsertRequest(tarantoolSpaceName).
Tuple([]interface{}{dir, nil, name, ttl, string(meta)}).
Operations(operations)
Operations(operations).
Context(ctx)
ret := crud.Result{}
@@ -178,7 +179,8 @@ func (store *TarantoolStore) FindEntry(ctx context.Context, fullpath weed_util.F
req := crud.NewGetRequest(tarantoolSpaceName).
Key([]interface{}{dir, name}).
Opts(findEntryGetOpts)
Opts(findEntryGetOpts).
Context(ctx)
resp := crud.Result{}
@@ -223,7 +225,8 @@ func (store *TarantoolStore) DeleteEntry(ctx context.Context, fullpath weed_util
req := crud.NewDeleteRequest(tarantoolSpaceName).
Key([]interface{}{dir, name}).
Opts(delOpts)
Opts(delOpts).
Context(ctx)
if _, err := store.pool.Do(req, pool.ModeRW).Get(); err != nil {
return fmt.Errorf("delete %s : %v", fullpath, err)
@@ -234,7 +237,8 @@ func (store *TarantoolStore) DeleteEntry(ctx context.Context, fullpath weed_util
func (store *TarantoolStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
req := tarantool.NewCallRequest("filer_metadata.delete_by_directory_idx").
Args([]interface{}{fullpath})
Args([]interface{}{fullpath}).
Context(ctx)
if _, err := store.pool.Do(req, pool.ModeRW).Get(); err != nil {
return fmt.Errorf("delete %s : %v", fullpath, err)
@@ -250,7 +254,8 @@ func (store *TarantoolStore) ListDirectoryPrefixedEntries(ctx context.Context, d
func (store *TarantoolStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
req := tarantool.NewCallRequest("filer_metadata.find_by_directory_idx_and_name").
Args([]interface{}{string(dirPath), startFileName, includeStartFile, limit})
Args([]interface{}{string(dirPath), startFileName, includeStartFile, limit}).
Context(ctx)
results, err := store.pool.Do(req, pool.ModePreferRO).Get()
if err != nil {
weed/filer/tarantool/tarantool_store_kv.go
+7 -4
View File
@@ -21,7 +21,7 @@ func (store *TarantoolStore) KvPut(ctx context.Context, key []byte, value []byte
var operations = []crud.Operation{
{
Operator: crud.Insert,
Operator: crud.Assign,
Field: "value",
Value: string(value),
},
@@ -29,7 +29,8 @@ func (store *TarantoolStore) KvPut(ctx context.Context, key []byte, value []byte
req := crud.NewUpsertRequest(tarantoolKVSpaceName).
Tuple([]interface{}{string(key), nil, string(value)}).
Operations(operations)
Operations(operations).
Context(ctx)
ret := crud.Result{}
if err := store.pool.Do(req, pool.ModeRW).GetTyped(&ret); err != nil {
@@ -50,7 +51,8 @@ func (store *TarantoolStore) KvGet(ctx context.Context, key []byte) (value []byt
req := crud.NewGetRequest(tarantoolKVSpaceName).
Key([]interface{}{string(key)}).
Opts(getOpts)
Opts(getOpts).
Context(ctx)
resp := crud.Result{}
@@ -85,7 +87,8 @@ func (store *TarantoolStore) KvDelete(ctx context.Context, key []byte) (err erro
req := crud.NewDeleteRequest(tarantoolKVSpaceName).
Key([]interface{}{string(key)}).
Opts(delOpts)
Opts(delOpts).
Context(ctx)
if _, err := store.pool.Do(req, pool.ModeRW).Get(); err != nil {
return fmt.Errorf("kv delete: %w", err)