[#192] Fix frostfs dependencies

Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>

api/data/info.go

@@ -59,6 +59,13 @@ type (
 	BucketSettings struct {
 		Versioning        string                   `json:"versioning"`
 		LockConfiguration *ObjectLockConfiguration `json:"lock_configuration"`
+		LifecycleConfig   *LifecycleConfig         `json:"lifecycle_configuration"`
+	}
+
+	// LifecycleConfig stores lifecycle config old and current settings.
+	LifecycleConfig struct {
+		OldConfigurationID   string `json:"old_id"`
+		CurrentConfiguration *LifecycleConfiguration
 	}
 
 	// CORSConfiguration stores CORS configuration of a request.
@@ -124,3 +131,6 @@ func (b BucketSettings) VersioningEnabled() bool {
 func (b BucketSettings) VersioningSuspended() bool {
 	return b.Versioning == VersioningSuspended
 }
+
+// ExpirationObject returns name of system object for expiration tick object.
+func (o *ObjectInfo) ExpirationObject() string { return ".expiration." + o.Name }

api/data/lifecycle.go

@@ -0,0 +1,171 @@
+package data
+
+import (
+	"encoding/xml"
+	"strings"
+)
+
+type (
+	LifecycleConfiguration struct {
+		XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ LifecycleConfiguration" json:"-"`
+		Rules   []Rule   `xml:"Rule" json:"Rule"`
+	}
+
+	Rule struct {
+		AbortIncompleteMultipartUpload *AbortIncompleteMultipartUpload `xml:"AbortIncompleteMultipartUpload" json:"AbortIncompleteMultipartUpload"`
+		Expiration                     *Expiration                     `xml:"Expiration" json:"Expiration"`
+		Filter                         *LifecycleRuleFilter            `xml:"Filter" json:"Filter"`
+		ID                             string                          `xml:"ID" json:"ID"`
+		NoncurrentVersionExpiration    *NoncurrentVersionExpiration    `xml:"NoncurrentVersionExpiration" json:"NoncurrentVersionExpiration"`
+		NoncurrentVersionTransitions   []NoncurrentVersionTransition   `xml:"NoncurrentVersionTransition" json:"NoncurrentVersionTransition"`
+		Prefix                         *string                         `xml:"Prefix" json:"Prefix"`
+		Status                         string                          `xml:"Status" json:"Status"`
+		Transitions                    []Transition                    `xml:"Transition" json:"Transition"`
+	}
+
+	AbortIncompleteMultipartUpload struct {
+		DaysAfterInitiation int64 `xml:"DaysAfterInitiation"`
+	}
+
+	Expiration struct {
+		Date                      *string `xml:"Date" json:"Date"`
+		Days                      *int64  `xml:"Days" json:"Days"`
+		ExpiredObjectDeleteMarker bool    `xml:"ExpiredObjectDeleteMarker" json:"ExpiredObjectDeleteMarker"`
+	}
+
+	LifecycleRuleFilter struct {
+		And                   *LifecycleRuleAndOperator `xml:"And" json:"And"`
+		ObjectSizeGreaterThan *int64                    `xml:"ObjectSizeGreaterThan" json:"ObjectSizeGreaterThan"`
+		ObjectSizeLessThan    *int64                    `xml:"ObjectSizeLessThan" json:"ObjectSizeLessThan"`
+		Prefix                *string                   `xml:"Prefix" json:"Prefix"`
+		Tag                   *Tag                      `xml:"Tag" json:"Tag"`
+	}
+
+	LifecycleRuleAndOperator struct {
+		ObjectSizeGreaterThan *int64  `xml:"ObjectSizeGreaterThan" json:"ObjectSizeGreaterThan"`
+		ObjectSizeLessThan    *int64  `xml:"ObjectSizeLessThan" json:"ObjectSizeLessThan"`
+		Prefix                *string `xml:"Prefix" json:"Prefix"`
+		Tags                  []Tag   `xml:"Tags" json:"Tags"`
+	}
+
+	Tag struct {
+		Key   string `xml:"Key" json:"Key"`
+		Value string `xml:"Value" json:"Value"`
+	}
+
+	NoncurrentVersionExpiration struct {
+		NewerNoncurrentVersions *int64 `xml:"NewerNoncurrentVersions" json:"NewerNoncurrentVersions"`
+		NoncurrentDays          *int64 `xml:"NoncurrentDays" json:"NoncurrentDays"`
+	}
+
+	NoncurrentVersionTransition struct {
+		NewerNoncurrentVersions *int64 `xml:"NewerNoncurrentVersions" json:"NewerNoncurrentVersions"`
+		NoncurrentDays          *int64 `xml:"NoncurrentDays" json:"NoncurrentDays"`
+		StorageClass            string `xml:"StorageClass" json:"StorageClass"`
+	}
+
+	Transition struct {
+		Date         *string `xml:"Date" json:"Date"`
+		Days         *int64  `xml:"Days" json:"Days"`
+		StorageClass string  `xml:"StorageClass" json:"StorageClass"`
+	}
+
+	ExpirationObject struct {
+		Expiration        *Expiration
+		RuleID            string
+		LifecycleConfigID string
+	}
+)
+
+func (r *Rule) RealPrefix() string {
+	if r.Filter == nil {
+		if r.Prefix != nil {
+			return *r.Prefix
+		}
+		return ""
+	}
+
+	if r.Filter.And == nil {
+		if r.Filter.Prefix != nil {
+			return *r.Filter.Prefix
+		}
+		return ""
+	}
+
+	if r.Filter.And.Prefix != nil {
+		return *r.Filter.And.Prefix
+	}
+	return ""
+}
+
+func (r *Rule) NeedTags() bool {
+	if r.Filter == nil {
+		return false
+	}
+
+	if r.Filter.And == nil {
+		return r.Filter.Tag != nil
+	}
+
+	return len(r.Filter.And.Tags) != 0
+}
+
+func (r *Rule) MatchObject(obj *ObjectInfo, tags map[string]string) bool {
+	if r.Filter == nil {
+		if r.Prefix != nil {
+			return strings.HasPrefix(obj.Name, *r.Prefix)
+		}
+		return true
+	}
+
+	if r.Filter.And == nil {
+		if r.Filter.Prefix != nil && !strings.HasPrefix(obj.Name, *r.Filter.Prefix) {
+			return false
+		}
+
+		if r.Filter.Tag != nil {
+			if tags == nil {
+				return false
+			}
+			if tagVal := tags[r.Filter.Tag.Key]; tagVal != r.Filter.Tag.Value {
+				return false
+			}
+		}
+
+		if r.Filter.ObjectSizeLessThan != nil && *r.Filter.ObjectSizeLessThan > 0 && obj.Size >= *r.Filter.ObjectSizeLessThan {
+			return false
+		}
+
+		if r.Filter.ObjectSizeGreaterThan != nil && obj.Size <= *r.Filter.ObjectSizeGreaterThan {
+			return false
+		}
+
+		return true
+	}
+
+	if r.Filter.And.Prefix != nil && !strings.HasPrefix(obj.Name, *r.Filter.And.Prefix) {
+		return false
+	}
+
+	if len(r.Filter.And.Tags) != 0 {
+		if tags == nil {
+			return false
+		}
+
+		for _, tag := range r.Filter.And.Tags {
+			if tagVal := tags[tag.Key]; tagVal != tag.Value {
+				return false
+			}
+		}
+	}
+
+	if r.Filter.And.ObjectSizeLessThan != nil && obj.Size >= *r.Filter.And.ObjectSizeLessThan {
+		return false
+	}
+
+	if r.Filter.And.ObjectSizeGreaterThan != nil && obj.Size <= *r.Filter.And.ObjectSizeGreaterThan {
+		return false
+	}
+
+	return true
+}

api/handler/get.go

@@ -195,6 +195,11 @@ func (h *handler) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if err = h.setExpirationHeader(r.Context(), bktInfo, info, w.Header()); err != nil {
+		h.logAndSendError(w, "could not get expiration info", reqInfo, err)
+		return
+	}
+
 	bktSettings, err := h.obj.GetBucketSettings(r.Context(), bktInfo)
 	if err != nil {
 		h.logAndSendError(w, "could not get bucket settings", reqInfo, err)

api/handler/head.go

@@ -2,7 +2,11 @@ package handler
 
 import (
 	"bytes"
+	"context"
+	"fmt"
 	"net/http"
+	"net/url"
+	"time"
 
 	"github.com/TrueCloudLab/frostfs-s3-gw/api"
 	"github.com/TrueCloudLab/frostfs-s3-gw/api/data"
@@ -103,6 +107,11 @@ func (h *handler) HeadObjectHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if err = h.setExpirationHeader(r.Context(), bktInfo, info, w.Header()); err != nil {
+		h.logAndSendError(w, "could not get expiration info", reqInfo, err)
+		return
+	}
+
 	bktSettings, err := h.obj.GetBucketSettings(r.Context(), bktInfo)
 	if err != nil {
 		h.logAndSendError(w, "could not get bucket settings", reqInfo, err)
@@ -157,3 +166,24 @@ func writeLockHeaders(h http.Header, legalHold *data.LegalHold, retention *data.
 		h.Set(api.AmzObjectLockMode, retention.Mode)
 	}
 }
+
+func (h *handler) setExpirationHeader(ctx context.Context, bktInfo *data.BucketInfo, objInfo *data.ObjectInfo, header http.Header) error {
+	var expirationObjInfo data.ObjectInfo
+
+	// todo get expiration object info
+
+	ruleID := expirationObjInfo.Headers[layer.AttributeExpireRuleID]
+
+	expDate, err := time.Parse(time.RFC3339, expirationObjInfo.Headers[layer.AttributeExpireDate])
+	if err != nil {
+		return fmt.Errorf("couldn't parse ivalid expiration time: %w", err)
+	}
+
+	writeExpirationHeader(header, ruleID, expDate)
+	return nil
+}
+
+func writeExpirationHeader(h http.Header, ruleID string, expDate time.Time) {
+	header := "expiry-date=\"%s\", rule-id=\"%s\""
+	h.Set(api.AmzExpiration, fmt.Sprintf(header, expDate.UTC().Format(http.TimeFormat), url.QueryEscape(ruleID)))
+}

api/handler/lifecycle.go

@@ -0,0 +1,168 @@
+package handler
+
+import (
+	"context"
+	"encoding/xml"
+	"fmt"
+	"net/http"
+
+	"github.com/TrueCloudLab/frostfs-s3-gw/api"
+	"github.com/TrueCloudLab/frostfs-s3-gw/api/data"
+	apiErrors "github.com/TrueCloudLab/frostfs-s3-gw/api/errors"
+)
+
+func (h *handler) PutBucketLifecycleHandler(w http.ResponseWriter, r *http.Request) {
+	reqInfo := api.GetReqInfo(r.Context())
+
+	bktInfo, err := h.obj.GetBucketInfo(r.Context(), reqInfo.BucketName)
+	if err != nil {
+		h.logAndSendError(w, "could not get bucket info", reqInfo, err)
+		return
+	}
+	if err = checkOwner(bktInfo, r.Header.Get(api.AmzExpectedBucketOwner)); err != nil {
+		h.logAndSendError(w, "expected owner doesn't match", reqInfo, err)
+		return
+	}
+
+	lifecycleConf := &data.LifecycleConfiguration{}
+	if err = xml.NewDecoder(r.Body).Decode(lifecycleConf); err != nil {
+		h.logAndSendError(w, "couldn't parse lifecycle configuration", reqInfo, err)
+		return
+	}
+
+	if err = checkLifecycleConfiguration(lifecycleConf); err != nil {
+		h.logAndSendError(w, "invalid lifecycle configuration", reqInfo, err)
+		return
+	}
+
+	if err = h.updateLifecycleConfiguration(r.Context(), bktInfo, lifecycleConf); err != nil {
+		h.logAndSendError(w, "couldn't put bucket settings", reqInfo, err)
+		return
+	}
+}
+
+func (h *handler) GetBucketLifecycleHandler(w http.ResponseWriter, r *http.Request) {
+	reqInfo := api.GetReqInfo(r.Context())
+
+	bktInfo, err := h.obj.GetBucketInfo(r.Context(), reqInfo.BucketName)
+	if err != nil {
+		h.logAndSendError(w, "could not get bucket info", reqInfo, err)
+		return
+	}
+	if err = checkOwner(bktInfo, r.Header.Get(api.AmzExpectedBucketOwner)); err != nil {
+		h.logAndSendError(w, "expected owner doesn't match", reqInfo, err)
+		return
+	}
+
+	settings, err := h.obj.GetBucketSettings(r.Context(), bktInfo)
+	if err != nil {
+		h.logAndSendError(w, "couldn't get bucket settings", reqInfo, err)
+		return
+	}
+
+	if settings.LifecycleConfig == nil || settings.LifecycleConfig.CurrentConfiguration == nil {
+		h.logAndSendError(w, "lifecycle configuration doesn't exist", reqInfo,
+			apiErrors.GetAPIError(apiErrors.ErrNoSuchLifecycleConfiguration))
+		return
+	}
+
+	if err = api.EncodeToResponse(w, settings.LifecycleConfig.CurrentConfiguration); err != nil {
+		h.logAndSendError(w, "something went wrong", reqInfo, err)
+	}
+}
+
+func (h *handler) DeleteBucketLifecycleHandler(w http.ResponseWriter, r *http.Request) {
+	reqInfo := api.GetReqInfo(r.Context())
+
+	bktInfo, err := h.obj.GetBucketInfo(r.Context(), reqInfo.BucketName)
+	if err != nil {
+		h.logAndSendError(w, "could not get bucket info", reqInfo, err)
+		return
+	}
+	if err = checkOwner(bktInfo, r.Header.Get(api.AmzExpectedBucketOwner)); err != nil {
+		h.logAndSendError(w, "expected owner doesn't match", reqInfo, err)
+		return
+	}
+
+	if err = h.updateLifecycleConfiguration(r.Context(), bktInfo, nil); err != nil {
+		h.logAndSendError(w, "couldn't put bucket settings", reqInfo, err)
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+func (h *handler) updateLifecycleConfiguration(ctx context.Context, bktInfo *data.BucketInfo, lifecycleConf *data.LifecycleConfiguration) error {
+	// todo consider run as separate goroutine
+	if err := h.obj.ScheduleLifecycle(ctx, bktInfo, lifecycleConf); err != nil {
+		return fmt.Errorf("couldn't apply lifecycle: %w", err)
+	}
+
+	return nil
+}
+
+func checkLifecycleConfiguration(conf *data.LifecycleConfiguration) error {
+	err := apiErrors.GetAPIError(apiErrors.ErrMalformedXML)
+
+	if len(conf.Rules) == 0 {
+		return err
+	}
+	if len(conf.Rules) > 1000 {
+		return fmt.Errorf("you cannot have more than 1000 rules")
+	}
+
+	for _, rule := range conf.Rules {
+		if rule.Status != enabledValue && rule.Status != disabledValue {
+			return err
+		}
+		if rule.Prefix != nil && rule.Filter != nil {
+			return err
+		}
+
+		if rule.Filter != nil {
+			if rule.Filter.ObjectSizeGreaterThan != nil && *rule.Filter.ObjectSizeGreaterThan < 0 ||
+				rule.Filter.ObjectSizeLessThan != nil && *rule.Filter.ObjectSizeLessThan < 0 {
+				return err
+			}
+
+			if !filterContainsOneOption(rule.Filter) {
+				return err
+			}
+		}
+
+		if !ruleHasAction(rule) {
+			return err
+		}
+
+		// currently only expiration action is supported
+		if rule.Expiration == nil {
+			return err
+		}
+		if rule.Expiration.Days != nil && rule.Expiration.Date != nil ||
+			rule.Expiration.Days == nil && rule.Expiration.Date == nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func filterContainsOneOption(filter *data.LifecycleRuleFilter) bool {
+	exactlyOneOption := 0
+	if filter.Prefix != nil {
+		exactlyOneOption++
+	}
+	if filter.And != nil {
+		exactlyOneOption++
+	}
+	if filter.Tag != nil {
+		exactlyOneOption++
+	}
+
+	return exactlyOneOption == 1
+}
+
+func ruleHasAction(rule data.Rule) bool {
+	return rule.AbortIncompleteMultipartUpload != nil || rule.Expiration != nil ||
+		rule.NoncurrentVersionExpiration != nil || len(rule.Transitions) != 0 ||
+		len(rule.NoncurrentVersionTransitions) != 0
+}

api/handler/lifecycle_test.go

@@ -0,0 +1,152 @@
+package handler
+
+import (
+	"encoding/xml"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"testing"
+
+	"github.com/TrueCloudLab/frostfs-s3-gw/api/data"
+	apiErrors "github.com/TrueCloudLab/frostfs-s3-gw/api/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCheckLifecycleConfiguration(t *testing.T) {
+	numRules := 1001
+	rules := make([]data.Rule, numRules)
+	for i := 0; i < numRules; i++ {
+		rules[i] = data.Rule{ID: strconv.Itoa(i), Status: disabledValue}
+	}
+
+	prefix := "prefix"
+	invalidSize := int64(-1)
+	days := int64(1)
+
+	for _, tc := range []struct {
+		name          string
+		configuration *data.LifecycleConfiguration
+		noError       bool
+	}{
+		{
+			name: "basic",
+			configuration: &data.LifecycleConfiguration{Rules: []data.Rule{{
+				ID:         "Some ID",
+				Status:     "Disabled",
+				Expiration: &data.Expiration{Days: &days},
+			}}},
+			noError: true,
+		},
+		{
+			name: "invalid status",
+			configuration: &data.LifecycleConfiguration{Rules: []data.Rule{{
+				ID:     "Some ID",
+				Status: "",
+			}}},
+		},
+		{
+			name:          "zero rules",
+			configuration: &data.LifecycleConfiguration{},
+		},
+		{
+			name:          "more than max rules",
+			configuration: &data.LifecycleConfiguration{Rules: rules},
+		},
+		{
+			name: "invalid empty filter",
+			configuration: &data.LifecycleConfiguration{Rules: []data.Rule{{
+				Status: enabledValue,
+				Filter: &data.LifecycleRuleFilter{},
+			}}},
+		},
+		{
+			name: "invalid filter not exactly one option",
+			configuration: &data.LifecycleConfiguration{Rules: []data.Rule{{
+				Status: enabledValue,
+				Filter: &data.LifecycleRuleFilter{
+					Prefix: &prefix,
+					Tag:    &data.Tag{},
+				},
+			}}},
+		},
+		{
+			name: "invalid filter greater obj size",
+			configuration: &data.LifecycleConfiguration{Rules: []data.Rule{{
+				Status: enabledValue,
+				Filter: &data.LifecycleRuleFilter{
+					ObjectSizeGreaterThan: &invalidSize,
+				},
+			}}},
+		},
+		{
+			name: "invalid filter less obj size",
+			configuration: &data.LifecycleConfiguration{Rules: []data.Rule{{
+				Status: enabledValue,
+				Filter: &data.LifecycleRuleFilter{
+					ObjectSizeLessThan: &invalidSize,
+				},
+			}}},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			err := checkLifecycleConfiguration(tc.configuration)
+			if tc.noError {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+			}
+		})
+	}
+}
+
+func TestBucketLifecycleConfiguration(t *testing.T) {
+	hc := prepareHandlerContext(t)
+
+	bktName := "bucket-for-lifecycle"
+	createTestBucket(hc, bktName)
+
+	w, r := prepareTestRequest(hc, bktName, "", nil)
+	hc.Handler().GetBucketLifecycleHandler(w, r)
+	assertS3Error(t, w, apiErrors.GetAPIError(apiErrors.ErrNoSuchLifecycleConfiguration))
+
+	days := int64(1)
+	lifecycleConf := &data.LifecycleConfiguration{
+		XMLName: xmlName("LifecycleConfiguration"),
+		Rules: []data.Rule{
+			{
+				Expiration: &data.Expiration{Days: &days},
+				ID:         "Test",
+				Status:     "Disabled",
+			},
+		}}
+	w, r = prepareTestRequest(hc, bktName, "", lifecycleConf)
+	hc.Handler().PutBucketLifecycleHandler(w, r)
+	require.Equal(t, http.StatusOK, w.Code)
+
+	w, r = prepareTestRequest(hc, bktName, "", nil)
+	hc.Handler().GetBucketLifecycleHandler(w, r)
+	assertXMLEqual(t, w, lifecycleConf, &data.LifecycleConfiguration{})
+
+	w, r = prepareTestRequest(hc, bktName, "", lifecycleConf)
+	hc.Handler().DeleteBucketLifecycleHandler(w, r)
+	require.Equal(t, http.StatusNoContent, w.Code)
+
+	// make sure deleting is idempotent operation
+	w, r = prepareTestRequest(hc, bktName, "", lifecycleConf)
+	hc.Handler().DeleteBucketLifecycleHandler(w, r)
+	require.Equal(t, http.StatusNoContent, w.Code)
+}
+
+func assertXMLEqual(t *testing.T, w *httptest.ResponseRecorder, expected, actual interface{}) {
+	err := xml.NewDecoder(w.Result().Body).Decode(actual)
+	require.NoError(t, err)
+	require.Equal(t, expected, actual)
+	require.Equal(t, http.StatusOK, w.Code)
+}
+
+func xmlName(local string) xml.Name {
+	return xml.Name{
+		Space: "http://s3.amazonaws.com/doc/2006-03-01/",
+		Local: local,
+	}
+}

api/handler/locking.go

@@ -19,6 +19,7 @@ const (
 	yearDuration = 365 * dayDuration
 
 	enabledValue   = "Enabled"
+	disabledValue  = "Disabled"
 	governanceMode = "GOVERNANCE"
 	complianceMode = "COMPLIANCE"
 	legalHoldOn    = "ON"

api/handler/not_support.go

@@ -11,10 +11,6 @@ func (h *handler) DeleteBucketPolicyHandler(w http.ResponseWriter, r *http.Reque
 	h.logAndSendError(w, "not supported", api.GetReqInfo(r.Context()), errors.GetAPIError(errors.ErrNotSupported))
 }
 
-func (h *handler) DeleteBucketLifecycleHandler(w http.ResponseWriter, r *http.Request) {
-	h.logAndSendError(w, "not supported", api.GetReqInfo(r.Context()), errors.GetAPIError(errors.ErrNotSupported))
-}
-
 func (h *handler) DeleteBucketEncryptionHandler(w http.ResponseWriter, r *http.Request) {
 	h.logAndSendError(w, "not supported", api.GetReqInfo(r.Context()), errors.GetAPIError(errors.ErrNotSupported))
 }

api/handler/unimplemented.go

@@ -11,10 +11,6 @@ func (h *handler) SelectObjectContentHandler(w http.ResponseWriter, r *http.Requ
 	h.logAndSendError(w, "not implemented", api.GetReqInfo(r.Context()), errors.GetAPIError(errors.ErrNotImplemented))
 }
 
-func (h *handler) GetBucketLifecycleHandler(w http.ResponseWriter, r *http.Request) {
-	h.logAndSendError(w, "not implemented", api.GetReqInfo(r.Context()), errors.GetAPIError(errors.ErrNotImplemented))
-}
-
 func (h *handler) GetBucketEncryptionHandler(w http.ResponseWriter, r *http.Request) {
 	h.logAndSendError(w, "not implemented", api.GetReqInfo(r.Context()), errors.GetAPIError(errors.ErrNotImplemented))
 }
@@ -51,10 +47,6 @@ func (h *handler) ListObjectsV2MHandler(w http.ResponseWriter, r *http.Request)
 	h.logAndSendError(w, "not implemented", api.GetReqInfo(r.Context()), errors.GetAPIError(errors.ErrNotImplemented))
 }
 
-func (h *handler) PutBucketLifecycleHandler(w http.ResponseWriter, r *http.Request) {
-	h.logAndSendError(w, "not implemented", api.GetReqInfo(r.Context()), errors.GetAPIError(errors.ErrNotImplemented))
-}
-
 func (h *handler) PutBucketEncryptionHandler(w http.ResponseWriter, r *http.Request) {
 	h.logAndSendError(w, "not implemented", api.GetReqInfo(r.Context()), errors.GetAPIError(errors.ErrNotImplemented))
 }

api/headers.go

@@ -57,6 +57,7 @@ const (
 	AmzObjectAttributes          = "X-Amz-Object-Attributes"
 	AmzMaxParts                  = "X-Amz-Max-Parts"
 	AmzPartNumberMarker          = "X-Amz-Part-Number-Marker"
+	AmzExpiration                = "X-Amz-Expiration"
 
 	AmzServerSideEncryptionCustomerAlgorithm = "x-amz-server-side-encryption-customer-algorithm"
 	AmzServerSideEncryptionCustomerKey       = "x-amz-server-side-encryption-customer-key"

api/layer/layer.go

@@ -191,6 +191,8 @@ type (
 		GetBucketSettings(ctx context.Context, bktInfo *data.BucketInfo) (*data.BucketSettings, error)
 		PutBucketSettings(ctx context.Context, p *PutSettingsParams) error
 
+		ScheduleLifecycle(ctx context.Context, bktInfo *data.BucketInfo, new *data.LifecycleConfiguration) error
+
 		PutBucketCORS(ctx context.Context, p *PutCORSParams) error
 		GetBucketCORS(ctx context.Context, bktInfo *data.BucketInfo) (*data.CORSConfiguration, error)
 		DeleteBucketCORS(ctx context.Context, bktInfo *data.BucketInfo) error
@@ -288,7 +290,9 @@ func (n *layer) Initialize(ctx context.Context, c EventListener) error {
 		return fmt.Errorf("already initialized")
 	}
 
-	// todo add notification handlers (e.g. for lifecycles)
+	if err := c.Subscribe(ctx, ExpireTopic, MsgHandlerFunc(n.handleExpireTick)); err != nil {
+		return fmt.Errorf("couldn't initialize layer: %w", err)
+	}
 
 	c.Listen(ctx)
 

api/layer/lifecycle.go

@@ -0,0 +1,211 @@
+package layer
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/xml"
+	"fmt"
+
+	"github.com/TrueCloudLab/frostfs-s3-gw/api/data"
+	oid "github.com/TrueCloudLab/frostfs-sdk-go/object/id"
+	"github.com/nats-io/nats.go"
+	"go.uber.org/zap"
+)
+
+const (
+	AttributeExpirationEpoch   = "__NEOFS__EXPIRATION_EPOCH"
+	AttributeSysTickEpoch      = "__NEOFS__TICK_EPOCH"
+	AttributeSysTickTopic      = "__NEOFS__TICK_TOPIC"
+	AttributeParentObject      = ".s3-expire-parent-object"
+	AttributeParentBucket      = ".s3-expire-parent-bucket"
+	AttributeExpireDate        = ".s3-expire-date"
+	AttributeExpireRuleID      = ".s3-expire-rule-id"
+	AttributeLifecycleConfigID = ".s3-lifecycle-config"
+	ExpireTopic                = "expire"
+)
+
+func (n *layer) handleExpireTick(ctx context.Context, msg *nats.Msg) error {
+	var addr oid.Address
+	if err := addr.DecodeString(string(msg.Data)); err != nil {
+		return fmt.Errorf("invalid msg, address expected: %w", err)
+	}
+
+	n.log.Debug("handling expiration tick", zap.String("address", string(msg.Data)))
+
+	// and make sure having right access
+
+	//todo redo
+	bktInfo := &data.BucketInfo{CID: addr.Container()}
+
+	obj, err := n.objectHead(ctx, bktInfo, addr.Object())
+	if err != nil {
+		return fmt.Errorf("couldn't head expiration object: %w", err)
+	}
+
+	header := userHeaders(obj.Attributes())
+	objName := header[AttributeParentObject]
+	bktName := header[AttributeParentBucket]
+	if objName == "" || bktName == "" {
+		return fmt.Errorf("couldn't know bucket/object to expire")
+	}
+
+	p := &DeleteObjectParams{
+		BktInfo: bktInfo,
+		Objects: []*VersionedObject{{Name: objName}},
+	}
+
+	res := n.DeleteObjects(ctx, p)
+	if res[0].Error != nil {
+		return fmt.Errorf("couldn't delete expired object: %w", res[0].Error)
+	}
+
+	return n.objectDelete(ctx, bktInfo, addr.Object())
+}
+
+func (n *layer) ScheduleLifecycle(ctx context.Context, bktInfo *data.BucketInfo, newConf *data.LifecycleConfiguration) error {
+	if newConf == nil {
+		return nil
+	}
+
+	lifecycleID, err := computeLifecycleID(newConf)
+	if err != nil {
+		return fmt.Errorf("couldn't compute lifecycle id: %w", err)
+	}
+
+	// We want to be able to revert partly applied lifecycle if something goes wrong.
+	if err = n.updateLifecycle(ctx, bktInfo, &data.LifecycleConfig{
+		OldConfigurationID: lifecycleID,
+	}); err != nil {
+		return err
+	}
+
+	if err = n.applyLifecycle(ctx, bktInfo, lifecycleID, newConf); err != nil {
+		return err
+	}
+
+	return n.updateLifecycle(ctx, bktInfo, &data.LifecycleConfig{
+		OldConfigurationID:   lifecycleID,
+		CurrentConfiguration: newConf,
+	})
+}
+
+func (n *layer) updateLifecycle(ctx context.Context, bktInfo *data.BucketInfo, lifecycleConfig *data.LifecycleConfig) error {
+	settings, err := n.GetBucketSettings(ctx, bktInfo)
+	if err != nil {
+		return fmt.Errorf("couldn't get bucket settings: %w", err)
+	}
+
+	settings.LifecycleConfig = lifecycleConfig
+	sp := &PutSettingsParams{
+		BktInfo:  bktInfo,
+		Settings: settings,
+	}
+
+	if err = n.PutBucketSettings(ctx, sp); err != nil {
+		return fmt.Errorf("couldn't put bucket settings: %w", err)
+	}
+	return nil
+}
+
+func (n *layer) applyLifecycle(ctx context.Context, bktInfo *data.BucketInfo, lifecycleID string, conf *data.LifecycleConfiguration) error {
+	for _, rule := range conf.Rules {
+		if rule.Status == "Disabled" {
+			continue
+		}
+
+		listParam := allObjectParams{
+			Bucket: bktInfo,
+			Prefix: rule.RealPrefix(),
+		}
+
+		objects, _, err := n.getLatestObjectsVersions(ctx, listParam)
+		if err != nil {
+			return err
+		}
+
+		if err = n.applyLifecycleToObjects(ctx, bktInfo, lifecycleID, rule, objects); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (n *layer) applyLifecycleToObjects(ctx context.Context, bktInfo *data.BucketInfo, lifecycleID string, rule data.Rule, objects []*data.ObjectInfo) error {
+	var tags []map[string]string
+	var err error
+	if rule.NeedTags() {
+		tags = make([]map[string]string, len(objects))
+		p := &ObjectVersion{
+			BktInfo: bktInfo,
+		}
+		for i, obj := range objects {
+			p.ObjectName = obj.Name
+			p.VersionID = obj.VersionID()
+			if _, tags[i], err = n.GetObjectTagging(ctx, p); err != nil {
+				return fmt.Errorf("couldn't get object tags: %w", err)
+			}
+		}
+	}
+
+	for i, obj := range objects {
+		var objTags map[string]string
+		if len(tags) != 0 {
+			objTags = tags[i]
+		}
+		if !rule.MatchObject(obj, objTags) {
+			continue
+		}
+
+		expObj := &data.ExpirationObject{
+			Expiration:        rule.Expiration,
+			RuleID:            rule.ID,
+			LifecycleConfigID: lifecycleID,
+		}
+
+		if _, err = n.putExpirationObject(ctx, bktInfo, obj, expObj); err != nil {
+			return fmt.Errorf("couldn't put expiration object: %w", err)
+		}
+	}
+
+	return nil
+}
+
+func (n *layer) putLifecycleObjects(ctx context.Context, bktInfo *data.BucketInfo, obj *data.ObjectInfo, lifecycle *data.LifecycleConfig) error {
+	if lifecycle == nil || lifecycle.CurrentConfiguration == nil {
+		return nil
+	}
+
+	for _, rule := range lifecycle.CurrentConfiguration.Rules {
+		if rule.Status == "Disabled" {
+			continue
+		}
+
+		// at this time lifecycle.OldConfigurationID is the same as lifecycle.CurrentConfiguration id
+		if err := n.applyLifecycleToObjects(ctx, bktInfo, lifecycle.OldConfigurationID, rule, []*data.ObjectInfo{obj}); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func computeLifecycleID(conf *data.LifecycleConfiguration) (string, error) {
+	raw, err := xml.Marshal(conf)
+	if err != nil {
+		return "", fmt.Errorf("couldn't marshall new lifecycle configuration: %w", err)
+	}
+
+	sha := sha256.New()
+	sha.Write(raw)
+	sum := sha.Sum(nil)
+
+	id := hex.EncodeToString(sum)
+
+	if id == "" {
+		return "", fmt.Errorf("computed id is empty")
+	}
+
+	return id, nil
+}

api/layer/lifecycle_test.go

@@ -0,0 +1,159 @@
+package layer
+
+import (
+	"testing"
+
+	"github.com/TrueCloudLab/frostfs-s3-gw/api/data"
+	"github.com/stretchr/testify/require"
+)
+
+func TestComputeLifecycleID(t *testing.T) {
+	conf := &data.LifecycleConfiguration{Rules: []data.Rule{
+		{
+			ID:     "id",
+			Status: "Enabled",
+		},
+	}}
+
+	id, err := computeLifecycleID(conf)
+	require.NoError(t, err)
+	require.Equal(t, "51ff619dc848622287764fc7c4aec06b7c1a5936c25b8eee48a0dbcb4eeac9f4", id)
+}
+
+func TestRuleMatchObject(t *testing.T) {
+	prefix, suffix := "prefix", "suffix"
+	objSizeMin, objSizeMax := int64(512), int64(1024)
+
+	for _, tc := range []struct {
+		name     string
+		rule     data.Rule
+		obj      *data.ObjectInfo
+		tags     map[string]string
+		expected bool
+	}{
+		{
+			name:     "basic match",
+			rule:     data.Rule{Prefix: &prefix},
+			obj:      &data.ObjectInfo{Name: prefix + suffix},
+			expected: true,
+		},
+		{
+			name:     "basic no match",
+			rule:     data.Rule{Prefix: &prefix},
+			obj:      &data.ObjectInfo{Name: suffix + prefix},
+			expected: false,
+		},
+		{
+			name: "filter and sizes",
+			rule: data.Rule{Filter: &data.LifecycleRuleFilter{
+				And: &data.LifecycleRuleAndOperator{
+					ObjectSizeGreaterThan: &objSizeMin,
+					ObjectSizeLessThan:    &objSizeMax,
+				},
+			}},
+			obj:      &data.ObjectInfo{Name: suffix, Size: 768},
+			expected: true,
+		},
+		{
+			name: "filter prefix",
+			rule: data.Rule{Filter: &data.LifecycleRuleFilter{
+				Prefix: &prefix,
+			}},
+			obj:      &data.ObjectInfo{Name: prefix + suffix},
+			expected: true,
+		},
+		{
+			name: "filter prefix no match",
+			rule: data.Rule{Filter: &data.LifecycleRuleFilter{
+				Prefix: &prefix,
+			}},
+			obj:      &data.ObjectInfo{Name: suffix},
+			expected: false,
+		},
+		{
+			name: "filter tags",
+			rule: data.Rule{Filter: &data.LifecycleRuleFilter{
+				Tag: &data.Tag{
+					Key:   "key",
+					Value: "val",
+				},
+			}},
+			tags:     map[string]string{"key": "val"},
+			obj:      &data.ObjectInfo{},
+			expected: true,
+		},
+		{
+			name: "filter and tags no match",
+			rule: data.Rule{Filter: &data.LifecycleRuleFilter{
+				And: &data.LifecycleRuleAndOperator{
+					Tags: []data.Tag{{
+						Key:   "key",
+						Value: "val",
+					}},
+				},
+			}},
+			tags:     map[string]string{"key": "val2"},
+			obj:      &data.ObjectInfo{},
+			expected: false,
+		},
+		{
+			name: "filter size no match",
+			rule: data.Rule{Filter: &data.LifecycleRuleFilter{
+				ObjectSizeGreaterThan: &objSizeMax,
+			}},
+			obj:      &data.ObjectInfo{Size: objSizeMin},
+			expected: false,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			require.Equal(t, tc.expected, tc.rule.MatchObject(tc.obj, tc.tags))
+		})
+	}
+}
+
+func TestScheduleLifecycle(t *testing.T) {
+	tc := prepareContext(t)
+
+	obj1 := tc.putObject([]byte("content"))
+
+	date := "2022-03-14T09:59:03Z"
+	date2 := "2022-03-15T09:59:03Z"
+	prefix := "prefix"
+	tc.obj = prefix
+	obj2 := tc.putObject([]byte("content2"))
+
+	conf := &data.LifecycleConfiguration{
+		Rules: []data.Rule{{
+			Filter: &data.LifecycleRuleFilter{
+				Prefix: &prefix,
+			},
+			Expiration: &data.Expiration{
+				Date: &date,
+			}},
+		},
+	}
+
+	err := tc.layer.ScheduleLifecycle(tc.ctx, tc.bktInfo, conf)
+	require.NoError(t, err)
+
+	expObj1, _ := tc.getObject(obj1.ExpirationObject(), "", false)
+	require.Nil(t, expObj1)
+	expObj2, _ := tc.getObject(obj2.ExpirationObject(), "", false)
+	require.NotNil(t, expObj2)
+	assertExpirationObject(t, expObj2, date)
+
+	conf.Rules[0].Expiration.Date = &date2
+	err = tc.layer.ScheduleLifecycle(tc.ctx, tc.bktInfo, conf)
+	require.NoError(t, err)
+
+	expObj2, _ = tc.getObject(obj2.ExpirationObject(), "", false)
+	require.NotNil(t, expObj2)
+	assertExpirationObject(t, expObj2, date2)
+}
+
+func assertExpirationObject(t *testing.T, expObjInfo *data.ObjectInfo, date string) {
+	require.Equal(t, expObjInfo.Headers[AttributeExpireDate], date)
+	require.Contains(t, expObjInfo.Headers, AttributeSysTickEpoch)
+	require.Contains(t, expObjInfo.Headers, AttributeSysTickTopic)
+	require.Contains(t, expObjInfo.Headers, AttributeLifecycleConfigID)
+}

api/layer/object.go

@@ -298,6 +298,11 @@ func (n *layer) PutObject(ctx context.Context, p *PutObjectParams) (*data.Extend
 		NodeVersion: newVersion,
 	}
 
+	// todo filling api.AmzExpiration header
+	if err = n.putLifecycleObjects(ctx, p.BktInfo, objInfo, bktSettings.LifecycleConfig); err != nil {
+		return nil, fmt.Errorf("couldn't put expiration system objects: %w", err)
+	}
+
 	n.cache.PutObjectWithName(owner, extendedObjInfo)
 
 	return extendedObjInfo, nil

api/layer/system_object.go

@@ -15,8 +15,7 @@ import (
 )
 
 const (
-	AttributeComplianceMode  = ".s3-compliance-mode"
-	AttributeExpirationEpoch = "__NEOFS__EXPIRATION_EPOCH"
+	AttributeComplianceMode = ".s3-compliance-mode"
 )
 
 type PutLockInfoParams struct {
@@ -130,6 +129,54 @@ func (n *layer) putLockObject(ctx context.Context, bktInfo *data.BucketInfo, obj
 	return id, err
 }
 
+func (n *layer) putExpirationObject(ctx context.Context, bktInfo *data.BucketInfo, objInfo *data.ObjectInfo, expObj *data.ExpirationObject) (oid.ID, error) {
+	prm := PrmObjectCreate{
+		Container: bktInfo.CID,
+		Creator:   bktInfo.Owner,
+		Filepath:  objInfo.ExpirationObject(),
+	}
+
+	var (
+		err     error
+		exp     uint64
+		expTime time.Time
+	)
+
+	if expObj.Expiration.Days != nil {
+		expTime = objInfo.Created.Add(time.Duration(*expObj.Expiration.Days) * 24 * time.Hour).UTC()
+		// emulate rounding the resulting time to the next day midnight UTC
+		toMidnight := 24 - expTime.UTC().Hour()
+		expTime = expTime.Add(time.Duration(toMidnight) * time.Hour)
+	} else {
+		expTime, err = time.Parse(time.RFC3339, *expObj.Expiration.Date)
+		if err != nil {
+			return oid.ID{}, fmt.Errorf("couldn't parse expiration date '%s': %w", *expObj.Expiration.Date, err)
+		}
+	}
+
+	now := TimeNow(ctx)
+	if expTime.After(now) {
+		_, exp, err = n.frostFS.TimeToEpoch(ctx, now, expTime)
+		if err != nil {
+			return oid.ID{}, fmt.Errorf("couldn't compute expiration epoch: %w", err)
+		}
+	}
+
+	prm.Attributes = [][2]string{
+		{AttributeExpirationEpoch, strconv.FormatUint(exp+4, 10)},
+		{AttributeSysTickEpoch, strconv.FormatUint(exp, 10)},
+		{AttributeSysTickTopic, ExpireTopic},
+		{AttributeParentObject, objInfo.Name},
+		{AttributeParentBucket, bktInfo.Name},
+		{AttributeExpireDate, expTime.Format(time.RFC3339)},
+		{AttributeExpireRuleID, expObj.RuleID},
+		{AttributeLifecycleConfigID, expObj.LifecycleConfigID},
+	}
+
+	id, _, err := n.objectPutAndHash(ctx, prm, bktInfo)
+	return id, err
+}
+
 func (n *layer) GetLockInfo(ctx context.Context, objVersion *ObjectVersion) (*data.LockInfo, error) {
 	owner := n.Owner(ctx)
 	if lockInfo := n.cache.GetLockInfo(owner, lockObjectKey(objVersion)); lockInfo != nil {

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/gorilla/mux v1.8.0
 	github.com/minio/sio v0.3.0
-	github.com/nats-io/nats.go v1.13.1-0.20220121202836-972a071d373d
+	github.com/nats-io/nats.go v1.19.0
 	github.com/nspcc-dev/neo-go v0.100.1
 	github.com/panjf2000/ants/v2 v2.5.0
 	github.com/prometheus/client_golang v1.13.0
@@ -19,7 +19,7 @@ require (
 	github.com/stretchr/testify v1.8.1
 	github.com/urfave/cli/v2 v2.3.0
 	go.uber.org/zap v1.24.0
-	golang.org/x/crypto v0.4.0
+	golang.org/x/crypto v0.5.0
 	google.golang.org/grpc v1.48.0
 	google.golang.org/protobuf v1.28.1
 )
@@ -39,7 +39,6 @@ require (
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
 	github.com/fsnotify/fsnotify v1.4.9 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/golang/snappy v0.0.3 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/hashicorp/golang-lru v0.6.0 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
@@ -48,7 +47,7 @@ require (
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/mitchellh/mapstructure v1.4.1 // indirect
 	github.com/mr-tron/base58 v1.2.0 // indirect
-	github.com/nats-io/nats-server/v2 v2.7.1 // indirect
+	github.com/nats-io/nats-server/v2 v2.9.11 // indirect
 	github.com/nats-io/nkeys v0.3.0 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/nspcc-dev/go-ordered-json v0.0.0-20220111165707-25110be27d22 // indirect
@@ -69,11 +68,11 @@ require (
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect
 	golang.org/x/exp v0.0.0-20221227203929-1b447090c38c // indirect
-	golang.org/x/net v0.3.0 // indirect
+	golang.org/x/net v0.5.0 // indirect
 	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4 // indirect
-	golang.org/x/sys v0.3.0 // indirect
-	golang.org/x/term v0.3.0 // indirect
-	golang.org/x/text v0.5.0 // indirect
+	golang.org/x/sys v0.4.0 // indirect
+	golang.org/x/term v0.4.0 // indirect
+	golang.org/x/text v0.6.0 // indirect
 	google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c // indirect
 	gopkg.in/ini.v1 v1.62.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect

go.sum

@@ -198,9 +198,8 @@ github.com/golang/protobuf v1.5.1/go.mod h1:DopwsBzvsk0Fs44TXzsVbJyPhcCPeIwnvohx
 github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
 github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
 github.com/golang/snappy v0.0.0-20170215233205-553a64147049/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
-github.com/golang/snappy v0.0.3 h1:fHPg5GQYlCeLIPB9BZqMVR5nR9A+IM5zcgeTdjMYmLA=
-github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/gomodule/redigo v2.0.0+incompatible/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
@@ -298,8 +297,8 @@ github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:C
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4=
-github.com/klauspost/compress v1.13.4 h1:0zhec2I8zGnjWcKyLl6i3gPqKANCCn5e9xmviEEeX6s=
-github.com/klauspost/compress v1.13.4/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
+github.com/klauspost/compress v1.15.11 h1:Lcadnb3RKGin4FYM/orgq0qde+nc15E5Cbqg4B9Sx9c=
+github.com/klauspost/compress v1.15.11/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
@@ -325,8 +324,8 @@ github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Ky
 github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
-github.com/minio/highwayhash v1.0.1 h1:dZ6IIu8Z14VlC0VpfKofAhCy74wu/Qb5gcn52yWoz/0=
-github.com/minio/highwayhash v1.0.1/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
+github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
+github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/minio/sio v0.3.0 h1:syEFBewzOMOYVzSTFpp1MqpSZk8rUNbz8VIIc+PNzus=
 github.com/minio/sio v0.3.0/go.mod h1:8b0yPp2avGThviy/+OCJBI6OMpvxoUuiLvE6F1lebhw=
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=
@@ -348,12 +347,12 @@ github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
 github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
-github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296 h1:vU9tpM3apjYlLLeY23zRWJ9Zktr5jp+mloR942LEOpY=
-github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
-github.com/nats-io/nats-server/v2 v2.7.1 h1:SDj8R0PJPVekw3EgHxGtTfJUuMbsuaul1nwWFI3xTyk=
-github.com/nats-io/nats-server/v2 v2.7.1/go.mod h1:tckmrt0M6bVaDT3kmh9UrIq/CBOBBse+TpXQi5ldaa8=
-github.com/nats-io/nats.go v1.13.1-0.20220121202836-972a071d373d h1:GRSmEJutHkdoxKsRypP575IIdoXe7Bm6yHQF6GcDBnA=
-github.com/nats-io/nats.go v1.13.1-0.20220121202836-972a071d373d/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/jwt/v2 v2.3.0 h1:z2mA1a7tIf5ShggOFlR1oBPgd6hGqcDYsISxZByUzdI=
+github.com/nats-io/jwt/v2 v2.3.0/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
+github.com/nats-io/nats-server/v2 v2.9.11 h1:4y5SwWvWI59V5mcqtuoqKq6L9NDUydOP3Ekwuwl8cZI=
+github.com/nats-io/nats-server/v2 v2.9.11/go.mod h1:b0oVuxSlkvS3ZjMkncFeACGyZohbO4XhSqW1Lt7iRRY=
+github.com/nats-io/nats.go v1.19.0 h1:H6j8aBnTQFoVrTGB6Xjd903UMdE7jz6DS4YkmAqgZ9Q=
+github.com/nats-io/nats.go v1.19.0/go.mod h1:tLqubohF7t4z3du1QDPYJIQQyhb4wl6DhjxEajSI7UA=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
@@ -413,6 +412,7 @@ github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZ
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
+github.com/prashantv/gostub v1.1.0/go.mod h1:A5zLQHz7ieHGG7is6LLXLz7I8+3LZzsrV0P1IAHhP5U=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
 github.com/prometheus/client_golang v1.2.1/go.mod h1:XMU6Z2MjaRKVu/dC1qupJI9SiNkDYzz3xecMgSW/F+U=
@@ -525,6 +525,7 @@ go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=
 go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
+go.uber.org/automaxprocs v1.5.1/go.mod h1:BF4eumQw0P9GtnuxxovUd06vwm1o18oMzFtK66vU6XU=
 go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
 go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI=
 go.uber.org/goleak v1.1.11/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
@@ -553,10 +554,10 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
-golang.org/x/crypto v0.4.0 h1:UVQgzMY87xqpKNgb+kDsll2Igd33HszWHFLmpaRMq/8=
 golang.org/x/crypto v0.4.0/go.mod h1:3quD/ATkf6oY+rnes5c3ExXTbLc8mueNue5/DoinL80=
+golang.org/x/crypto v0.5.0 h1:U/0M97KRkSFvyD/3FSmdP5W5swImpNgle/EHFhOsQPE=
+golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
@@ -643,13 +644,13 @@ golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLd
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
-golang.org/x/net v0.3.0 h1:VWL6FNY2bEEmsGVKabSlHu5Irp34xmMRoqb/9lF9lxk=
 golang.org/x/net v0.3.0/go.mod h1:MBQ8lrhLObU/6UmLb4fmbmk5OcyYmqtbGd/9yIeKjEE=
+golang.org/x/net v0.5.0 h1:GyT4nK/YDHSqa1c4753ouYCDajOYKTja9Xb/OHtgvSw=
+golang.org/x/net v0.5.0/go.mod h1:DivGGAXEgPSlEBzxGzZI+ZLohi+xUj054jfeKui00ws=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -744,19 +745,20 @@ golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211019181941-9d821ace8654/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220111092808-5a964db01320/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.3.0 h1:w8ZOecv6NaNa/zC8944JTU3vz4u6Lagfk4RPQxv92NQ=
 golang.org/x/sys v0.3.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.4.0 h1:Zr2JFtRQNX3BCZ8YtxRE9hNJYC8J6I1MVbMg6owUp18=
+golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210429154555-c04ba851c2a4/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
-golang.org/x/term v0.3.0 h1:qoo4akIqOcDME5bhc/NgxUdovd6BSS2uMsVjB56q1xI=
 golang.org/x/term v0.3.0/go.mod h1:q750SLmJuPmVoN1blW3UFBPREJfb1KmY3vwxfr+nFDA=
+golang.org/x/term v0.4.0 h1:O7UWfv5+A2qiuulQk30kVinPoMtoIPeVaKLEgLpVkvg=
+golang.org/x/term v0.4.0/go.mod h1:9P2UbLfCdcvo3p/nzKvsmas4TnlujnuoV9hGgYzW1lQ=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -767,13 +769,14 @@ golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
-golang.org/x/text v0.5.0 h1:OLmvp0KP+FVG99Ct/qFiL/Fhk4zp4QQnZ7b2U+5piUM=
 golang.org/x/text v0.5.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.6.0 h1:3XmdazWV+ubf7QgHSTWeykHOci5oeekaGJBLkrkaw4k=
+golang.org/x/text v0.6.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
-golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 h1:GZokNIeuVkl3aZHJchRrr13WCsols02MLUcz1U9is6M=
-golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20220922220347-f3bd1da661af h1:Yx9k8YCG3dvF87UAn2tu2HQLf2dt/eR1bXxpLMWeH+Y=
+golang.org/x/time v0.0.0-20220922220347-f3bd1da661af/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180318012157-96caea41033d/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=