package layer import ( "bytes" "context" "fmt" "strconv" "sync" "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/data" "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer/frostfs" "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/middleware" "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/logs" objectV2 "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/object" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client" cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/relations" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user" "go.uber.org/zap" ) func (n *Layer) putTombstones(ctx context.Context, bkt *data.BucketInfo, networkInfo netmap.NetworkInfo, members []oid.ID) error { if len(members) == 0 { return nil } var wg sync.WaitGroup tombstoneMembersSize := n.features.TombstoneMembersSize() tombstoneLifetime := n.features.TombstoneLifetime() tombstonesCount := len(members) / tombstoneMembersSize if len(members)%tombstoneMembersSize != 0 { tombstonesCount++ } errCh := make(chan error, tombstonesCount) for i := 0; i < tombstonesCount; i++ { end := tombstoneMembersSize * (i + 1) if end > len(members) { end = len(members) } n.submitPutTombstone(ctx, bkt, members[tombstoneMembersSize*i:end], networkInfo.CurrentEpoch()+tombstoneLifetime, &wg, errCh) } wg.Wait() close(errCh) if err := <-errCh; err != nil { return err } return nil } func (n *Layer) submitPutTombstone(ctx context.Context, bkt *data.BucketInfo, members []oid.ID, expEpoch uint64, wg *sync.WaitGroup, errCh chan<- error) { tomb := object.NewTombstone() tomb.SetExpirationEpoch(expEpoch) tomb.SetMembers(members) wg.Add(1) err := n.workerPool.Submit(func() { defer wg.Done() if err := n.putTombstoneObject(ctx, tomb, bkt); err != nil { n.reqLogger(ctx).Warn(logs.FailedToPutTombstoneObject, zap.String("cid", bkt.CID.EncodeToString()), zap.Error(err)) errCh <- fmt.Errorf("put tombstone object: %w", err) } }) if err != nil { wg.Done() n.reqLogger(ctx).Warn(logs.FailedToSubmitTaskToPool, zap.Error(err)) errCh <- fmt.Errorf("submit task to pool: %w", err) } } func (n *Layer) putTombstoneObject(ctx context.Context, tomb *object.Tombstone, bktInfo *data.BucketInfo) error { payload, err := tomb.Marshal() if err != nil { return fmt.Errorf("marshal tombstone: %w", err) } prm := frostfs.PrmObjectCreate{ Container: bktInfo.CID, Attributes: [][2]string{{objectV2.SysAttributeExpEpoch, strconv.FormatUint(tomb.ExpirationEpoch(), 10)}}, Payload: bytes.NewReader(payload), CreationTime: TimeNow(ctx), ClientCut: n.features.ClientCut(), WithoutHomomorphicHash: bktInfo.HomomorphicHashDisabled, BufferMaxSize: n.features.BufferMaxSizeForPut(), Type: object.TypeTombstone, } n.prepareAuthParameters(ctx, &prm.PrmAuth, bktInfo.Owner) _, err = n.frostFS.CreateObject(ctx, prm) return err } func (n *Layer) getMembers(ctx context.Context, cnrID cid.ID, objID oid.ID, tokens relations.Tokens) ([]oid.ID, error) { oids, err := relations.ListAllRelations(ctx, n.frostFS.Relations(), cnrID, objID, tokens) if err != nil { if !client.IsErrObjectAlreadyRemoved(err) && !client.IsErrObjectNotFound(err) { return nil, fmt.Errorf("failed to list all object relations '%s': %w", objID.EncodeToString(), err) } n.reqLogger(ctx).Warn(logs.FailedToListAllObjectRelations, zap.String("cid", cnrID.EncodeToString()), zap.String("oid", objID.EncodeToString()), zap.Error(err)) return nil, nil } return append(oids, objID), nil } func prepareTokensParameter(ctx context.Context, bktOwner user.ID) relations.Tokens { tokens := relations.Tokens{} if bd, err := middleware.GetBoxData(ctx); err == nil && bd.Gate.BearerToken != nil { if bd.Gate.BearerToken.Impersonate() || bktOwner.Equals(bearer.ResolveIssuer(*bd.Gate.BearerToken)) { tokens.Bearer = bd.Gate.BearerToken } } return tokens }