package meta import ( "context" "errors" "fmt" "strconv" "time" objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/metaerr" "git.frostfs.info/TrueCloudLab/frostfs-observability/tracing" cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" "go.etcd.io/bbolt" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" ) // FilterExpired return expired items from addresses. // Address considered expired if metabase does contain information about expiration and // expiration epoch is less than epoch. func (db *DB) FilterExpired(ctx context.Context, epoch uint64, addresses []oid.Address) ([]oid.Address, error) { var ( startedAt = time.Now() success = true ) defer func() { db.metrics.AddMethodDuration("FilterExpired", time.Since(startedAt), success) }() _, span := tracing.StartSpanFromContext(ctx, "metabase.FilterExpired", trace.WithAttributes( attribute.String("epoch", strconv.FormatUint(epoch, 10)), attribute.Int("addr_count", len(addresses)), )) defer span.End() db.modeMtx.RLock() defer db.modeMtx.RUnlock() if db.mode.NoMetabase() { return nil, ErrDegradedMode } result := make([]oid.Address, 0, len(addresses)) containerIDToObjectIDs := make(map[cid.ID][]oid.ID) for _, addr := range addresses { containerIDToObjectIDs[addr.Container()] = append(containerIDToObjectIDs[addr.Container()], addr.Object()) } err := db.boltDB.View(func(tx *bbolt.Tx) error { for containerID, objectIDs := range containerIDToObjectIDs { select { case <-ctx.Done(): return ErrInterruptIterator default: } expiredNeoFS, err := selectExpiredObjectIDs(tx, objectV2.SysAttributeExpEpochNeoFS, epoch, containerID, objectIDs) if err != nil { return err } expiredSys, err := selectExpiredObjectIDs(tx, objectV2.SysAttributeExpEpoch, epoch, containerID, objectIDs) if err != nil { return err } for _, o := range expiredNeoFS { var a oid.Address a.SetContainer(containerID) a.SetObject(o) result = append(result, a) } for _, o := range expiredSys { var a oid.Address a.SetContainer(containerID) a.SetObject(o) result = append(result, a) } } return nil }) if err != nil { return nil, metaerr.Wrap(err) } success = true return result, nil } func isExpiredWithAttribute(tx *bbolt.Tx, attr string, addr oid.Address, currEpoch uint64) bool { // bucket with objects that have expiration attr attrKey := make([]byte, bucketKeySize+len(attr)) expirationBucket := tx.Bucket(attributeBucketName(addr.Container(), attr, attrKey)) if expirationBucket != nil { // bucket that contains objects that expire in the current epoch prevEpochBkt := expirationBucket.Bucket([]byte(strconv.FormatUint(currEpoch-1, 10))) if prevEpochBkt != nil { rawOID := objectKey(addr.Object(), make([]byte, objectKeySize)) if prevEpochBkt.Get(rawOID) != nil { return true } } } return false } func selectExpiredObjectIDs(tx *bbolt.Tx, attr string, epoch uint64, containerID cid.ID, objectIDs []oid.ID) ([]oid.ID, error) { result := make([]oid.ID, 0) notResolved := make(map[oid.ID]struct{}) for _, oid := range objectIDs { notResolved[oid] = struct{}{} } expiredBuffer := make([]oid.ID, 0) objectKeyBuffer := make([]byte, objectKeySize) expirationBucketKey := make([]byte, bucketKeySize+len(attr)) expirationBucket := tx.Bucket(attributeBucketName(containerID, attr, expirationBucketKey)) if expirationBucket == nil { return result, nil // all not expired } err := expirationBucket.ForEach(func(epochExpBucketKey, _ []byte) error { bucketExpiresAfter, err := strconv.ParseUint(string(epochExpBucketKey), 10, 64) if err != nil { return fmt.Errorf("could not parse expiration epoch: %w", err) } else if bucketExpiresAfter >= epoch { return nil } epochExpirationBucket := expirationBucket.Bucket(epochExpBucketKey) if epochExpirationBucket == nil { return nil } expiredBuffer = expiredBuffer[:0] for oid := range notResolved { key := objectKey(oid, objectKeyBuffer) if epochExpirationBucket.Get(key) != nil { expiredBuffer = append(expiredBuffer, oid) } } for _, oid := range expiredBuffer { delete(notResolved, oid) result = append(result, oid) } if len(notResolved) == 0 { return errBreakBucketForEach } return nil }) if err != nil && !errors.Is(err, errBreakBucketForEach) { return nil, err } return result, nil }