Evgenii Stratonikov
b42bcdc6fa
All checks were successful
DCO action / DCO (pull_request) Successful in 1m24s
Tests and linters / Run gofumpt (pull_request) Successful in 1m18s
Vulncheck / Vulncheck (pull_request) Successful in 1m30s
Pre-commit hooks / Pre-commit (pull_request) Successful in 2m20s
Build / Build Components (pull_request) Successful in 2m30s
Tests and linters / gopls check (pull_request) Successful in 2m50s
Tests and linters / Staticcheck (pull_request) Successful in 3m22s
Tests and linters / Lint (pull_request) Successful in 3m55s
Tests and linters / Tests (pull_request) Successful in 6m3s
Tests and linters / Tests with -race (pull_request) Successful in 6m2s
Consider the following operations ordering: 1. Inhume(with tombstone A) --> add tombstone mark for an object 2. --> new epoch arives 3. --> GCMark is added for a tombstone A, because it is unavailable 4. Put(A) --> return error, because the object already has a GCMark It is possible, and I have successfully reproduced it with a test on the shard level. However, the error is related to the specific _ordering_ of operations with engine. And triggering race-conditions like this is only possible on a shard level currently, so no tests are written. Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
55 lines
1.8 KiB
Go
55 lines
1.8 KiB
Go
package writer
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
containerCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container"
|
|
objectCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
|
|
containerSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
|
|
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
|
)
|
|
|
|
// ObjectStorage is an object storage interface.
|
|
type ObjectStorage interface {
|
|
// Put must save passed object
|
|
// and return any appeared error.
|
|
Put(context.Context, *objectSDK.Object, bool) error
|
|
// Delete must delete passed objects
|
|
// and return any appeared error.
|
|
Delete(ctx context.Context, tombstone oid.Address, toDelete []oid.ID) error
|
|
// Lock must lock passed objects
|
|
// and return any appeared error.
|
|
Lock(ctx context.Context, locker oid.Address, toLock []oid.ID) error
|
|
// IsLocked must clarify object's lock status.
|
|
IsLocked(context.Context, oid.Address) (bool, error)
|
|
}
|
|
|
|
type LocalTarget struct {
|
|
Storage ObjectStorage
|
|
Container containerSDK.Container
|
|
}
|
|
|
|
func (t LocalTarget) WriteObject(ctx context.Context, obj *objectSDK.Object, meta objectCore.ContentMeta) error {
|
|
if err := t.Storage.Put(ctx, obj, containerCore.IsIndexedContainer(t.Container)); err != nil {
|
|
return fmt.Errorf("(%T) could not put object to local storage: %w", t, err)
|
|
}
|
|
|
|
switch meta.Type() {
|
|
case objectSDK.TypeTombstone:
|
|
err := t.Storage.Delete(ctx, objectCore.AddressOf(obj), meta.Objects())
|
|
if err != nil {
|
|
return fmt.Errorf("could not delete objects from tombstone locally: %w", err)
|
|
}
|
|
case objectSDK.TypeLock:
|
|
err := t.Storage.Lock(ctx, objectCore.AddressOf(obj), meta.Objects())
|
|
if err != nil {
|
|
return fmt.Errorf("could not lock object from lock objects locally: %w", err)
|
|
}
|
|
default:
|
|
// objects that do not change meta storage
|
|
}
|
|
|
|
return nil
|
|
}
|