From a3afe4460b67184e70353df0d8df166c5eb68db2 Mon Sep 17 00:00:00 2001 From: MaratKarimov Date: Tue, 25 Aug 2026 01:17:08 +0300 Subject: [PATCH] tarantool: fix upsert data corruption and missing context propagation (#10926) Co-authored-by: Marat Karimov --- weed/filer/tarantool/tarantool_store.go | 17 +++++++++++------ weed/filer/tarantool/tarantool_store_kv.go | 11 +++++++---- 2 files changed, 18 insertions(+), 10 deletions(-) diff --git a/weed/filer/tarantool/tarantool_store.go b/weed/filer/tarantool/tarantool_store.go index 5094c8e24..54a6647b2 100644 --- a/weed/filer/tarantool/tarantool_store.go +++ b/weed/filer/tarantool/tarantool_store.go @@ -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 { diff --git a/weed/filer/tarantool/tarantool_store_kv.go b/weed/filer/tarantool/tarantool_store_kv.go index 4123cb2ab..6819360ba 100644 --- a/weed/filer/tarantool/tarantool_store_kv.go +++ b/weed/filer/tarantool/tarantool_store_kv.go @@ -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)