From fbd47fe1736a4336bae15cc4a40ecb1e462ad112 Mon Sep 17 00:00:00 2001 From: Piotr Piotrowski Date: Thu, 24 Apr 2025 14:18:47 +0200 Subject: [PATCH 1/3] [ADDED] KeyValue per key TTL support and limit markers Signed-off-by: Piotr Piotrowski --- go_test.mod | 5 +- go_test.sum | 12 +-- jetstream/errors.go | 16 ++++ jetstream/jetstream.go | 6 ++ jetstream/kv.go | 192 ++++++++++++++++++++++++++++---------- jetstream/kv_options.go | 28 ++++++ jetstream/message.go | 3 + jetstream/test/kv_test.go | 161 ++++++++++++++++++++++++++++++++ js.go | 2 +- test/kv_test.go | 2 +- 10 files changed, 366 insertions(+), 61 deletions(-) diff --git a/go_test.mod b/go_test.mod index 66a91fb38..37e67a598 100644 --- a/go_test.mod +++ b/go_test.mod @@ -6,7 +6,7 @@ require ( github.com/golang/protobuf v1.4.2 github.com/klauspost/compress v1.18.0 github.com/nats-io/jwt v1.2.2 - github.com/nats-io/nats-server/v2 v2.11.1 + github.com/nats-io/nats-server/v2 v2.11.2-RC.2.0.20250424120456-d24d162bc1f2 github.com/nats-io/nkeys v0.4.11 github.com/nats-io/nuid v1.0.1 go.uber.org/goleak v1.3.0 @@ -16,8 +16,7 @@ require ( require ( github.com/google/go-tpm v0.9.3 // indirect github.com/minio/highwayhash v1.0.3 // indirect - github.com/nats-io/jwt/v2 v2.7.3 // indirect - go.uber.org/automaxprocs v1.6.0 // indirect + github.com/nats-io/jwt/v2 v2.7.4 // indirect golang.org/x/crypto v0.37.0 // indirect golang.org/x/sys v0.32.0 // indirect golang.org/x/time v0.11.0 // indirect diff --git a/go_test.sum b/go_test.sum index 9792fb9fe..e352e8789 100644 --- a/go_test.sum +++ b/go_test.sum @@ -22,12 +22,10 @@ github.com/minio/highwayhash v1.0.3 h1:kbnuUMoHYyVl7szWjSxJnxw11k2U709jqFPPmIUyD github.com/minio/highwayhash v1.0.3/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= github.com/nats-io/jwt v1.2.2 h1:w3GMTO969dFg+UOKTmmyuu7IGdusK+7Ytlt//OYH/uU= github.com/nats-io/jwt v1.2.2/go.mod h1:/xX356yQA6LuXI9xWW7mZNpxgF2mBmGecH+Fj34sP5Q= -github.com/nats-io/jwt/v2 v2.7.3 h1:6bNPK+FXgBeAqdj4cYQ0F8ViHRbi7woQLq4W29nUAzE= -github.com/nats-io/jwt/v2 v2.7.3/go.mod h1:GvkcbHhKquj3pkioy5put1wvPxs78UlZ7D/pY+BgZk4= -github.com/nats-io/nats-server/v2 v2.11.0 h1:fdwAT1d6DZW/4LUz5rkvQUe5leGEwjjOQYntzVRKvjE= -github.com/nats-io/nats-server/v2 v2.11.0/go.mod h1:leXySghbdtXSUmWem8K9McnJ6xbJOb0t9+NQ5HTRZjI= -github.com/nats-io/nats-server/v2 v2.11.1 h1:LwdauqMqMNhTxTN3+WFTX6wGDOKntHljgZ+7gL5HCnk= -github.com/nats-io/nats-server/v2 v2.11.1/go.mod h1:leXySghbdtXSUmWem8K9McnJ6xbJOb0t9+NQ5HTRZjI= +github.com/nats-io/jwt/v2 v2.7.4 h1:jXFuDDxs/GQjGDZGhNgH4tXzSUK6WQi2rsj4xmsNOtI= +github.com/nats-io/jwt/v2 v2.7.4/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA= +github.com/nats-io/nats-server/v2 v2.11.2-RC.2.0.20250424120456-d24d162bc1f2 h1:2NxjLkGtoi4rpVDVI5Cl4JodbS3o3zwDudEqiy3+AYQ= +github.com/nats-io/nats-server/v2 v2.11.2-RC.2.0.20250424120456-d24d162bc1f2/go.mod h1:6Z6Fd+JgckqzKig7DYwhgrE7bJ6fypPHnGPND+DqgMY= github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s= github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0= github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE= @@ -37,8 +35,6 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= -go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs= -go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= diff --git a/jetstream/errors.go b/jetstream/errors.go index 5e50f488e..49a21adf5 100644 --- a/jetstream/errors.go +++ b/jetstream/errors.go @@ -330,6 +330,22 @@ var ( // ErrNoKeysFound is returned when no keys are found. ErrNoKeysFound JetStreamError = &jsError{message: "no keys found"} + // ErrInvalidLimitMarkerTTL is returned when an invalid value for + // LimitMarkerTTL is set when creating Key Value bucket. + ErrInvalidLimitMarkerTTL JetStreamError = &jsError{message: "limit marker TTL must be at least 1 second"} + + // ErrTTLOnDeleteNotSupported is returned when attempting to set a TTL + // on a delete operation. + ErrTTLOnDeleteNotSupported JetStreamError = &jsError{message: "TTL is not supported on delete"} + + // ErrLimitMarkerTTLNotSupported is returned when the connected jetstream API + // does not support setting the LimitMarkerTTL. + ErrLimitMarkerTTLNotSupported JetStreamError = &jsError{message: "limit marker TTLs not supported by server"} + + // ErrLimitMarkersNotEnabled is returned when attempting to set a ttl on a + // create or purge operation without enabling LimitMarkers on a bucket. + ErrLimitMarkersNotEnabled JetStreamError = &jsError{message: "limit markers need to be enabled to use this feature"} + // ErrObjectConfigRequired is returned when attempting to create an object // without a config. ErrObjectConfigRequired JetStreamError = &jsError{message: "object-store config required"} diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go index a99c2a0c3..960febeb0 100644 --- a/jetstream/jetstream.go +++ b/jetstream/jetstream.go @@ -261,11 +261,17 @@ type ( // APIStats reports on API calls to JetStream for this account. APIStats struct { + // Level is the API level for this account. + Level int `json:"level"` + // Total is the total number of API calls. Total uint64 `json:"total"` // Errors is the total number of API errors. Errors uint64 `json:"errors"` + + // Inflight is the number of API calls currently in flight. + Inflight uint64 `json:"inflight,omitempty"` } // AccountLimits includes the JetStream limits of the current account. diff --git a/jetstream/kv.go b/jetstream/kv.go index c4994629b..d215f38ba 100644 --- a/jetstream/kv.go +++ b/jetstream/kv.go @@ -119,7 +119,7 @@ type ( // // A key has to consist of alphanumeric characters, dashes, underscores, // equal signs, and dots. - Create(ctx context.Context, key string, value []byte) (uint64, error) + Create(ctx context.Context, key string, value []byte, opts ...KVCreateOpt) (uint64, error) // Update will update the value if the latest revision matches. // If the provided revision is not the latest, Update will return an error. @@ -251,6 +251,10 @@ type ( // Compression sets the underlying stream compression. // NOTE: Compression is supported for nats-server 2.10.0+ Compression bool `json:"compression,omitempty"` + + // LimitMarkerTTL is how long the bucket keeps markers when keys are + // removed by the TTL setting. + LimitMarkerTTL time.Duration } // KeyLister is used to retrieve a list of key value store keys. It returns @@ -306,6 +310,10 @@ type ( // IsCompressed indicates if the data is compressed on disk. IsCompressed() bool + + // LimitMarkerTTL is how long the bucket keeps markers when keys are + // removed by the TTL setting, 0 meaning markers are not supported. + LimitMarkerTTL() time.Duration } // KeyWatcher is what is returned when doing a watch. It can be used to @@ -375,6 +383,18 @@ type ( // Delete only if the latest revision matches. revision uint64 + + // purge ttl + ttl time.Duration + } + + // KVCreateOpt is used to configure Create. + KVCreateOpt interface { + configureCreate(opts *createOpts) error + } + + createOpts struct { + ttl time.Duration // TTL for the key } // KVPurgeOpt is used to configure PurgeDeletes. @@ -402,6 +422,8 @@ type kvs struct { useJSPfx bool // To know if we can use the stream direct get API useDirect bool + // To know whether KV uses limit markers and per msg ttl + useLimitMarkers bool } // KeyValueOp represents the type of KV operation (Put, Delete, Purge). It is a @@ -611,25 +633,43 @@ func (js *jetStream) prepareKeyValueConfig(ctx context.Context, cfg KeyValueConf if cfg.Compression { compression = S2Compression } + var allowMsgTTL bool + var subjectDeleteMarkerTTL time.Duration + if cfg.LimitMarkerTTL != 0 { + if cfg.LimitMarkerTTL < time.Second { + return StreamConfig{}, ErrInvalidLimitMarkerTTL + } + info, err := js.AccountInfo(ctx) + if err != nil { + return StreamConfig{}, err + } + if info.API.Level < 1 { + return StreamConfig{}, ErrLimitMarkerTTLNotSupported + } + allowMsgTTL = true + subjectDeleteMarkerTTL = cfg.LimitMarkerTTL + } scfg := StreamConfig{ - Name: fmt.Sprintf(kvBucketNameTmpl, cfg.Bucket), - Description: cfg.Description, - MaxMsgsPerSubject: history, - MaxBytes: maxBytes, - MaxAge: cfg.TTL, - MaxMsgSize: maxMsgSize, - Storage: cfg.Storage, - Replicas: replicas, - Placement: cfg.Placement, - AllowRollup: true, - DenyDelete: true, - Duplicates: duplicateWindow, - MaxMsgs: -1, - MaxConsumers: -1, - AllowDirect: true, - RePublish: cfg.RePublish, - Compression: compression, - Discard: DiscardNew, + Name: fmt.Sprintf(kvBucketNameTmpl, cfg.Bucket), + Description: cfg.Description, + MaxMsgsPerSubject: history, + MaxBytes: maxBytes, + MaxAge: cfg.TTL, + MaxMsgSize: maxMsgSize, + Storage: cfg.Storage, + Replicas: replicas, + Placement: cfg.Placement, + AllowRollup: true, + DenyDelete: true, + Duplicates: duplicateWindow, + MaxMsgs: -1, + MaxConsumers: -1, + AllowDirect: true, + RePublish: cfg.RePublish, + Compression: compression, + Discard: DiscardNew, + AllowMsgTTL: allowMsgTTL, + SubjectDeleteMarkerTTL: subjectDeleteMarkerTTL, } if cfg.Mirror != nil { // Copy in case we need to make changes so we do not change caller's version. @@ -730,7 +770,7 @@ func (js *jetStream) KeyValueStores(ctx context.Context) KeyValueLister { if !strings.HasPrefix(info.Config.Name, kvBucketNamePre) { continue } - res.kvs <- &KeyValueBucketStatus{nfo: info, bucket: strings.TrimPrefix(info.Config.Name, kvBucketNamePre)} + res.kvs <- &KeyValueBucketStatus{info: info, bucket: strings.TrimPrefix(info.Config.Name, kvBucketNamePre)} } if errors.Is(err, ErrEndOfData) { return @@ -742,7 +782,7 @@ func (js *jetStream) KeyValueStores(ctx context.Context) KeyValueLister { // KeyValueBucketStatus represents status of a Bucket, implements KeyValueStatus type KeyValueBucketStatus struct { - nfo *StreamInfo + info *StreamInfo bucket string } @@ -750,25 +790,31 @@ type KeyValueBucketStatus struct { func (s *KeyValueBucketStatus) Bucket() string { return s.bucket } // Values is how many messages are in the bucket, including historical values -func (s *KeyValueBucketStatus) Values() uint64 { return s.nfo.State.Msgs } +func (s *KeyValueBucketStatus) Values() uint64 { return s.info.State.Msgs } // History returns the configured history kept per key -func (s *KeyValueBucketStatus) History() int64 { return s.nfo.Config.MaxMsgsPerSubject } +func (s *KeyValueBucketStatus) History() int64 { return s.info.Config.MaxMsgsPerSubject } // TTL is how long the bucket keeps values for -func (s *KeyValueBucketStatus) TTL() time.Duration { return s.nfo.Config.MaxAge } +func (s *KeyValueBucketStatus) TTL() time.Duration { return s.info.Config.MaxAge } // BackingStore indicates what technology is used for storage of the bucket func (s *KeyValueBucketStatus) BackingStore() string { return "JetStream" } // StreamInfo is the stream info retrieved to create the status -func (s *KeyValueBucketStatus) StreamInfo() *StreamInfo { return s.nfo } +func (s *KeyValueBucketStatus) StreamInfo() *StreamInfo { return s.info } // Bytes is the size of the stream -func (s *KeyValueBucketStatus) Bytes() uint64 { return s.nfo.State.Bytes } +func (s *KeyValueBucketStatus) Bytes() uint64 { return s.info.State.Bytes } // IsCompressed indicates if the data is compressed on disk -func (s *KeyValueBucketStatus) IsCompressed() bool { return s.nfo.Config.Compression != NoCompression } +func (s *KeyValueBucketStatus) IsCompressed() bool { return s.info.Config.Compression != NoCompression } + +// LimitMarkerTTL is how long the bucket keeps markers when keys are +// removed by the TTL setting, 0 meaning markers are not supported. +func (s *KeyValueBucketStatus) LimitMarkerTTL() time.Duration { + return s.info.Config.SubjectDeleteMarkerTTL +} type kvLister struct { kvs chan KeyValueStatus @@ -862,12 +908,22 @@ func (kv *kvs) get(ctx context.Context, key string, revision uint64) (KeyValueEn // Double check here that this is not a DEL Operation marker. if len(m.Header) > 0 { - switch m.Header.Get(kvop) { - case kvdel: - entry.op = KeyValueDelete - return entry, ErrKeyDeleted - case kvpurge: - entry.op = KeyValuePurge + if m.Header.Get(kvop) != "" { + switch m.Header.Get(kvop) { + case kvdel: + entry.op = KeyValueDelete + case kvpurge: + entry.op = KeyValuePurge + } + } else if m.Header.Get(MarkerReasonHeader) != "" { + switch m.Header.Get(MarkerReasonHeader) { + case "MaxAge", "Purge": + entry.op = KeyValuePurge + case "Remove": + entry.op = KeyValueDelete + } + } + if entry.op != KeyValuePut { return entry, ErrKeyDeleted } } @@ -949,15 +1005,27 @@ func (kv *kvs) PutString(ctx context.Context, key string, value string) (uint64, return kv.Put(ctx, key, []byte(value)) } -// Create will add the key/value pair iff it does not exist. -func (kv *kvs) Create(ctx context.Context, key string, value []byte) (revision uint64, err error) { - v, err := kv.Update(ctx, key, value, 0) +// Create will add the key/value pair if it does not exist. +func (kv *kvs) Create(ctx context.Context, key string, value []byte, opts ...KVCreateOpt) (revision uint64, err error) { + var o createOpts + for _, opt := range opts { + if opt != nil { + if err := opt.configureCreate(&o); err != nil { + return 0, err + } + } + } + + if o.ttl > 0 && !kv.useLimitMarkers { + return 0, ErrLimitMarkersNotEnabled + } + v, err := kv.updateRevision(ctx, key, value, 0, o.ttl) if err == nil { return v, nil } if e, err := kv.get(ctx, key, kvLatestRevision); errors.Is(err, ErrKeyDeleted) { - return kv.Update(ctx, key, value, e.Revision()) + return kv.updateRevision(ctx, key, value, e.Revision(), o.ttl) } // Check if the expected last subject sequence is not zero which implies @@ -972,6 +1040,10 @@ func (kv *kvs) Create(ctx context.Context, key string, value []byte) (revision u // Update will update the value if the latest revision matches. func (kv *kvs) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + return kv.updateRevision(ctx, key, value, revision, 0) +} + +func (kv *kvs) updateRevision(ctx context.Context, key string, value []byte, revision uint64, ttl time.Duration) (uint64, error) { if !keyValid(key) { return 0, ErrInvalidKey } @@ -984,9 +1056,14 @@ func (kv *kvs) Update(ctx context.Context, key string, value []byte, revision ui b.WriteString(key) m := nats.Msg{Subject: b.String(), Header: nats.Header{}, Data: value} - m.Header.Set(ExpectedLastSubjSeqHeader, strconv.FormatUint(revision, 10)) + opts := []PublishOpt{ + WithExpectLastSequencePerSubject(revision), + } + if ttl > 0 { + opts = append(opts, WithMsgTTL(ttl)) + } - pa, err := kv.js.PublishMsg(ctx, &m) + pa, err := kv.js.PublishMsg(ctx, &m, opts...) if err != nil { return 0, err } @@ -1028,12 +1105,21 @@ func (kv *kvs) Delete(ctx context.Context, key string, opts ...KVDeleteOpt) erro } else { m.Header.Set(kvop, kvdel) } + pubOpts := make([]PublishOpt, 0) + if o.ttl > 0 && o.purge { + if !kv.useLimitMarkers { + return ErrLimitMarkersNotEnabled + } + pubOpts = append(pubOpts, WithMsgTTL(o.ttl)) + } else if o.ttl > 0 { + return ErrTTLOnDeleteNotSupported + } if o.revision != 0 { m.Header.Set(ExpectedLastSubjSeqHeader, strconv.FormatUint(o.revision, 10)) } - _, err := kv.js.PublishMsg(ctx, m) + _, err := kv.js.PublishMsg(ctx, m, pubOpts...) return err } @@ -1122,11 +1208,20 @@ func (kv *kvs) WatchFiltered(ctx context.Context, keys []string, opts ...WatchOp var op KeyValueOp if len(m.Header) > 0 { - switch m.Header.Get(kvop) { - case kvdel: - op = KeyValueDelete - case kvpurge: - op = KeyValuePurge + if m.Header.Get(kvop) != "" { + switch m.Header.Get(kvop) { + case kvdel: + op = KeyValueDelete + case kvpurge: + op = KeyValuePurge + } + } else if m.Header.Get(MarkerReasonHeader) != "" { + switch m.Header.Get(MarkerReasonHeader) { + case "MaxAge", "Purge": + op = KeyValuePurge + case "Remove": + op = KeyValueDelete + } } } delta := parser.ParseNum(tokens[parser.AckNumPendingTokenPos]) @@ -1404,7 +1499,7 @@ func (kv *kvs) Status(ctx context.Context) (KeyValueStatus, error) { return nil, err } - return &KeyValueBucketStatus{nfo: nfo, bucket: kv.name}, nil + return &KeyValueBucketStatus{info: nfo, bucket: kv.name}, nil } func mapStreamToKVS(js *jetStream, pushJS nats.JetStreamContext, stream Stream) *kvs { @@ -1418,8 +1513,9 @@ func mapStreamToKVS(js *jetStream, pushJS nats.JetStreamContext, stream Stream) pushJS: pushJS, stream: stream, // Determine if we need to use the JS prefix in front of Put and Delete operations - useJSPfx: js.opts.apiPrefix != DefaultAPIPrefix, - useDirect: info.Config.AllowDirect, + useJSPfx: js.opts.apiPrefix != DefaultAPIPrefix, + useDirect: info.Config.AllowDirect, + useLimitMarkers: info.Config.AllowMsgTTL, } // If we are mirroring, we will have mirror direct on, so just use the mirror name diff --git a/jetstream/kv_options.go b/jetstream/kv_options.go index 07a255727..8699c8802 100644 --- a/jetstream/kv_options.go +++ b/jetstream/kv_options.go @@ -101,3 +101,31 @@ func LastRevision(revision uint64) KVDeleteOpt { return nil }) } + +// PurgeTTL sets the TTL for the purge operation. +// After the TTL expires, the delete markers will be removed. +// This requires LimitMarkerTTL to be enabled on the bucket. +// Note that this is not the same as the TTL for the key itself, which is set +// using the KeyTTL option when creating the key. +func PurgeTTL(ttl time.Duration) KVDeleteOpt { + return deleteOptFn(func(opts *deleteOpts) error { + opts.ttl = ttl + return nil + }) +} + +type createOptFn func(opts *createOpts) error + +func (opt createOptFn) configureCreate(opts *createOpts) error { + return opt(opts) +} + +// KeyTTL sets the TTL for the key. This is the time after which the key will be +// automatically deleted. The TTL is set when the key is created and can not be +// changed later. This requires LimitMarkerTTL to be enabled on the bucket. +func KeyTTL(ttl time.Duration) KVCreateOpt { + return createOptFn(func(opts *createOpts) error { + opts.ttl = ttl + return nil + }) +} diff --git a/jetstream/message.go b/jetstream/message.go index 9103c7457..c21091b59 100644 --- a/jetstream/message.go +++ b/jetstream/message.go @@ -198,6 +198,9 @@ const ( // MsgRollup is used to apply a purge of all prior messages in the stream // ("all") or at the subject ("sub") before this message. MsgRollup = "Nats-Rollup" + + // MarkerReasonHeader is used to specify a reason for message deletion. + MarkerReasonHeader = "Nats-Marker-Reason" ) // Headers for republished messages and direct gets. Those headers are set by diff --git a/jetstream/test/kv_test.go b/jetstream/test/kv_test.go index 08e654c54..3e6ab63bf 100644 --- a/jetstream/test/kv_test.go +++ b/jetstream/test/kv_test.go @@ -1902,3 +1902,164 @@ func TestKeyValueCreateRepairOldKV(t *testing.T) { t.Fatalf("Expected error to be ErrBucketExists, got: %v", err) } } + +func TestKeyValueLimitMarkerTTL(t *testing.T) { + checkMsgHeaders := func(t *testing.T, js jetstream.JetStream, kv jetstream.KeyValue, key, expectedTTL, expectedReason string) { + t.Helper() + ctx := context.Background() + stream, err := js.Stream(ctx, "KV_KVS") + expectOk(t, err) + msg, err := stream.GetLastMsgForSubject(ctx, "$KV.KVS."+key) + expectOk(t, err) + marker := msg.Header.Get(jetstream.MarkerReasonHeader) + if marker != expectedReason { + t.Fatalf("Expected marker to be MaxAge, got %q", marker) + } + ttl := msg.Header.Get(jetstream.MsgTTLHeader) + if ttl != expectedTTL { + t.Fatalf("Expected TTL to be 1s, got %q", ttl) + } + } + + checkMsgNotFound := func(t *testing.T, js jetstream.JetStream, kv jetstream.KeyValue, key string) { + t.Helper() + ctx := context.Background() + stream, err := js.Stream(ctx, "KV_KVS") + expectOk(t, err) + msg, err := stream.GetLastMsgForSubject(ctx, "$KV.KVS."+key) + if err == nil { + t.Fatalf("Expected error getting message, got %v", msg) + } + if !errors.Is(err, jetstream.ErrMsgNotFound) { + t.Fatalf("Expected error to be ErrMsgNotFound, got: %v", err) + } + } + + t.Run("create with TTL", func(t *testing.T) { + s := RunBasicJetStreamServer() + defer shutdownJSServerAndRemoveStorage(t, s) + + nc, js := jsClient(t, s) + defer nc.Close() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", LimitMarkerTTL: time.Second}) + expectOk(t, err) + + // Put in a few names and ages. + _, err = kv.Create(ctx, "age", []byte("22"), jetstream.KeyTTL(time.Second)) + expectOk(t, err) + + // create watcher to wait for deletion + watcher, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + expectOk(t, err) + + _, err = kv.Get(ctx, "age") + expectOk(t, err) + time.Sleep(1500 * time.Millisecond) + + _, err = kv.Get(ctx, "age") + expectErr(t, err, jetstream.ErrKeyNotFound) + // check if marker exists on stream + checkMsgHeaders(t, js, kv, "age", "1s", "MaxAge") + + time.Sleep(time.Second) + _, err = kv.Get(ctx, "age") + expectErr(t, err, jetstream.ErrKeyNotFound) + // now msg should be gone from stream + checkMsgNotFound(t, js, kv, "age") + + entry := <-watcher.Updates() + if entry == nil { + t.Fatalf("Expected entry, got nil") + } + if entry.Operation() != jetstream.KeyValuePurge { + t.Fatalf("Expected purge operation, got %v", entry.Operation()) + } + if entry.Key() != "age" { + t.Fatalf("Expected key %q, got %q", "age", entry.Key()) + } + }) + + t.Run("purge with TTL", func(t *testing.T) { + s := RunBasicJetStreamServer() + defer shutdownJSServerAndRemoveStorage(t, s) + + nc, js := jsClient(t, s) + defer nc.Close() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", LimitMarkerTTL: time.Second}) + expectOk(t, err) + + // Put in a few names and ages. + _, err = kv.Create(ctx, "age", []byte("22")) + expectOk(t, err) + + watcher, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + expectOk(t, err) + + err = kv.Purge(ctx, "age", jetstream.PurgeTTL(time.Second)) + expectOk(t, err) + + _, err = kv.Get(ctx, "age") + expectErr(t, err, jetstream.ErrKeyNotFound) + // check if msg with ttl exists on stream + checkMsgHeaders(t, js, kv, "age", "1s", "") + expectErr(t, err, jetstream.ErrKeyNotFound) + + time.Sleep(1500 * time.Millisecond) + _, err = kv.Get(ctx, "age") + expectErr(t, err, jetstream.ErrKeyNotFound) + + // check if marker exists on stream + checkMsgHeaders(t, js, kv, "age", "1s", "MaxAge") + time.Sleep(time.Second) + // now msg should be gone from stream + checkMsgNotFound(t, js, kv, "age") + + entry := <-watcher.Updates() + if entry == nil { + t.Fatalf("Expected entry, got nil") + } + if entry.Operation() != jetstream.KeyValuePurge { + t.Fatalf("Expected purge operation, got %v", entry.Operation()) + } + if entry.Key() != "age" { + t.Fatalf("Expected key %q, got %q", "age", entry.Key()) + } + }) + + t.Run("invalid options", func(t *testing.T) { + s := RunBasicJetStreamServer() + defer shutdownJSServerAndRemoveStorage(t, s) + + nc, js := jsClient(t, s) + defer nc.Close() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", LimitMarkerTTL: 100 * time.Millisecond}) + expectErr(t, err, jetstream.ErrInvalidLimitMarkerTTL) + + // now create a kv without LimitMarkerTTL + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS"}) + expectOk(t, err) + + _, err = kv.Create(ctx, "age", []byte("22"), jetstream.KeyTTL(time.Second)) + expectErr(t, err, jetstream.ErrLimitMarkersNotEnabled) + + err = kv.Purge(ctx, "age", jetstream.PurgeTTL(time.Second)) + expectErr(t, err, jetstream.ErrLimitMarkersNotEnabled) + + // update kv to enable LimitMarkerTTL + kv, err = js.UpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", LimitMarkerTTL: time.Second}) + expectOk(t, err) + + // create a value with a TTL + _, err = kv.Create(ctx, "age", []byte("22"), jetstream.KeyTTL(time.Second)) + expectOk(t, err) + }) +} diff --git a/js.go b/js.go index 132e69ad6..6dafb8c45 100644 --- a/js.go +++ b/js.go @@ -348,7 +348,7 @@ type featureFlags struct { } // UseLegacyDurableConsumers makes JetStream use the legacy (pre nats-server v2.9.0) subjects for consumer creation. -// If this option is used when creating JetStremContext, $JS.API.CONSUMER.DURABLE.CREATE.. will be used +// If this option is used when creating JetStreamContext, $JS.API.CONSUMER.DURABLE.CREATE.. will be used // to create a consumer with Durable provided, rather than $JS.API.CONSUMER.CREATE... func UseLegacyDurableConsumers() JSOpt { return jsOptFn(func(opts *jsOpts) error { diff --git a/test/kv_test.go b/test/kv_test.go index 13c09ffd3..3fce540be 100644 --- a/test/kv_test.go +++ b/test/kv_test.go @@ -1390,7 +1390,7 @@ func TestKeyValueNonDirectGet(t *testing.T) { t.Fatalf("Error on get: v=%+v err=%v", v, err) } if v, err := kvi.GetRevision("key1", 1); err != nil || string(v.Value()) != "val1" { - t.Fatalf("Error on get revisiong: v=%+v err=%v", v, err) + t.Fatalf("Error on get revision: v=%+v err=%v", v, err) } if v, err := kvi.GetRevision("key1", 2); err == nil { t.Fatalf("Expected error, got %+v", v) From 6e83df1b458ef1a15b83c3edf3f46ddc15c3cdf2 Mon Sep 17 00:00:00 2001 From: Piotr Piotrowski Date: Mon, 28 Apr 2025 10:38:57 +0200 Subject: [PATCH 2/3] Remove goleak --- go_test.mod | 3 +-- go_test.sum | 14 ++------------ jetstream/test/main_test.go | 24 ------------------------ micro/test/main_test.go | 24 ------------------------ test/main_test.go | 24 ------------------------ 5 files changed, 3 insertions(+), 86 deletions(-) delete mode 100644 jetstream/test/main_test.go delete mode 100644 micro/test/main_test.go delete mode 100644 test/main_test.go diff --git a/go_test.mod b/go_test.mod index 37e67a598..5d2111b24 100644 --- a/go_test.mod +++ b/go_test.mod @@ -6,10 +6,9 @@ require ( github.com/golang/protobuf v1.4.2 github.com/klauspost/compress v1.18.0 github.com/nats-io/jwt v1.2.2 - github.com/nats-io/nats-server/v2 v2.11.2-RC.2.0.20250424120456-d24d162bc1f2 + github.com/nats-io/nats-server/v2 v2.11.2 github.com/nats-io/nkeys v0.4.11 github.com/nats-io/nuid v1.0.1 - go.uber.org/goleak v1.3.0 google.golang.org/protobuf v1.23.0 ) diff --git a/go_test.sum b/go_test.sum index e352e8789..6d411a20a 100644 --- a/go_test.sum +++ b/go_test.sum @@ -1,7 +1,5 @@ github.com/antithesishq/antithesis-sdk-go v0.4.3-default-no-op h1:+OSa/t11TFhqfrX0EOSqQBDJ0YlpmK0rDSiB19dg9M0= github.com/antithesishq/antithesis-sdk-go v0.4.3-default-no-op/go.mod h1:IUpT2DPAKh6i/YhSbt6Gl3v2yvUZjmKncl7U91fup7E= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= -github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8= github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA= github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs= @@ -24,19 +22,13 @@ github.com/nats-io/jwt v1.2.2 h1:w3GMTO969dFg+UOKTmmyuu7IGdusK+7Ytlt//OYH/uU= github.com/nats-io/jwt v1.2.2/go.mod h1:/xX356yQA6LuXI9xWW7mZNpxgF2mBmGecH+Fj34sP5Q= github.com/nats-io/jwt/v2 v2.7.4 h1:jXFuDDxs/GQjGDZGhNgH4tXzSUK6WQi2rsj4xmsNOtI= github.com/nats-io/jwt/v2 v2.7.4/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA= -github.com/nats-io/nats-server/v2 v2.11.2-RC.2.0.20250424120456-d24d162bc1f2 h1:2NxjLkGtoi4rpVDVI5Cl4JodbS3o3zwDudEqiy3+AYQ= -github.com/nats-io/nats-server/v2 v2.11.2-RC.2.0.20250424120456-d24d162bc1f2/go.mod h1:6Z6Fd+JgckqzKig7DYwhgrE7bJ6fypPHnGPND+DqgMY= +github.com/nats-io/nats-server/v2 v2.11.2 h1:k5KBAuRpJW9qAF11Io2txNhR5m1KUmqVkalLAw2yLfk= +github.com/nats-io/nats-server/v2 v2.11.2/go.mod h1:6Z6Fd+JgckqzKig7DYwhgrE7bJ6fypPHnGPND+DqgMY= github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s= github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0= github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE= github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= -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/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk= -github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= -go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= -go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE= @@ -58,5 +50,3 @@ google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miE google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo= google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM= google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= -gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= -gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/jetstream/test/main_test.go b/jetstream/test/main_test.go deleted file mode 100644 index d24136542..000000000 --- a/jetstream/test/main_test.go +++ /dev/null @@ -1,24 +0,0 @@ -// Copyright 2023 The NATS Authors -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package test - -import ( - "testing" - - "go.uber.org/goleak" -) - -func TestMain(m *testing.M) { - goleak.VerifyTestMain(m) -} diff --git a/micro/test/main_test.go b/micro/test/main_test.go deleted file mode 100644 index ceda72918..000000000 --- a/micro/test/main_test.go +++ /dev/null @@ -1,24 +0,0 @@ -// Copyright 2023 The NATS Authors -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package micro_test - -import ( - "testing" - - "go.uber.org/goleak" -) - -func TestMain(m *testing.M) { - goleak.VerifyTestMain(m) -} diff --git a/test/main_test.go b/test/main_test.go deleted file mode 100644 index d24136542..000000000 --- a/test/main_test.go +++ /dev/null @@ -1,24 +0,0 @@ -// Copyright 2023 The NATS Authors -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package test - -import ( - "testing" - - "go.uber.org/goleak" -) - -func TestMain(m *testing.M) { - goleak.VerifyTestMain(m) -} From 2127c16a631972b37da7fe43514bc5df29f634f7 Mon Sep 17 00:00:00 2001 From: Piotr Piotrowski Date: Wed, 30 Apr 2025 11:41:06 +0200 Subject: [PATCH 3/3] Remove client side validation Signed-off-by: Piotr Piotrowski --- jetstream/errors.go | 8 -------- jetstream/kv.go | 16 ++-------------- jetstream/test/kv_test.go | 31 ------------------------------- 3 files changed, 2 insertions(+), 53 deletions(-) diff --git a/jetstream/errors.go b/jetstream/errors.go index 49a21adf5..c7d06588f 100644 --- a/jetstream/errors.go +++ b/jetstream/errors.go @@ -330,10 +330,6 @@ var ( // ErrNoKeysFound is returned when no keys are found. ErrNoKeysFound JetStreamError = &jsError{message: "no keys found"} - // ErrInvalidLimitMarkerTTL is returned when an invalid value for - // LimitMarkerTTL is set when creating Key Value bucket. - ErrInvalidLimitMarkerTTL JetStreamError = &jsError{message: "limit marker TTL must be at least 1 second"} - // ErrTTLOnDeleteNotSupported is returned when attempting to set a TTL // on a delete operation. ErrTTLOnDeleteNotSupported JetStreamError = &jsError{message: "TTL is not supported on delete"} @@ -342,10 +338,6 @@ var ( // does not support setting the LimitMarkerTTL. ErrLimitMarkerTTLNotSupported JetStreamError = &jsError{message: "limit marker TTLs not supported by server"} - // ErrLimitMarkersNotEnabled is returned when attempting to set a ttl on a - // create or purge operation without enabling LimitMarkers on a bucket. - ErrLimitMarkersNotEnabled JetStreamError = &jsError{message: "limit markers need to be enabled to use this feature"} - // ErrObjectConfigRequired is returned when attempting to create an object // without a config. ErrObjectConfigRequired JetStreamError = &jsError{message: "object-store config required"} diff --git a/jetstream/kv.go b/jetstream/kv.go index d215f38ba..f2c140502 100644 --- a/jetstream/kv.go +++ b/jetstream/kv.go @@ -422,8 +422,6 @@ type kvs struct { useJSPfx bool // To know if we can use the stream direct get API useDirect bool - // To know whether KV uses limit markers and per msg ttl - useLimitMarkers bool } // KeyValueOp represents the type of KV operation (Put, Delete, Purge). It is a @@ -636,9 +634,6 @@ func (js *jetStream) prepareKeyValueConfig(ctx context.Context, cfg KeyValueConf var allowMsgTTL bool var subjectDeleteMarkerTTL time.Duration if cfg.LimitMarkerTTL != 0 { - if cfg.LimitMarkerTTL < time.Second { - return StreamConfig{}, ErrInvalidLimitMarkerTTL - } info, err := js.AccountInfo(ctx) if err != nil { return StreamConfig{}, err @@ -1016,9 +1011,6 @@ func (kv *kvs) Create(ctx context.Context, key string, value []byte, opts ...KVC } } - if o.ttl > 0 && !kv.useLimitMarkers { - return 0, ErrLimitMarkersNotEnabled - } v, err := kv.updateRevision(ctx, key, value, 0, o.ttl) if err == nil { return v, nil @@ -1107,9 +1099,6 @@ func (kv *kvs) Delete(ctx context.Context, key string, opts ...KVDeleteOpt) erro } pubOpts := make([]PublishOpt, 0) if o.ttl > 0 && o.purge { - if !kv.useLimitMarkers { - return ErrLimitMarkersNotEnabled - } pubOpts = append(pubOpts, WithMsgTTL(o.ttl)) } else if o.ttl > 0 { return ErrTTLOnDeleteNotSupported @@ -1513,9 +1502,8 @@ func mapStreamToKVS(js *jetStream, pushJS nats.JetStreamContext, stream Stream) pushJS: pushJS, stream: stream, // Determine if we need to use the JS prefix in front of Put and Delete operations - useJSPfx: js.opts.apiPrefix != DefaultAPIPrefix, - useDirect: info.Config.AllowDirect, - useLimitMarkers: info.Config.AllowMsgTTL, + useJSPfx: js.opts.apiPrefix != DefaultAPIPrefix, + useDirect: info.Config.AllowDirect, } // If we are mirroring, we will have mirror direct on, so just use the mirror name diff --git a/jetstream/test/kv_test.go b/jetstream/test/kv_test.go index 3e6ab63bf..4856aad80 100644 --- a/jetstream/test/kv_test.go +++ b/jetstream/test/kv_test.go @@ -2031,35 +2031,4 @@ func TestKeyValueLimitMarkerTTL(t *testing.T) { t.Fatalf("Expected key %q, got %q", "age", entry.Key()) } }) - - t.Run("invalid options", func(t *testing.T) { - s := RunBasicJetStreamServer() - defer shutdownJSServerAndRemoveStorage(t, s) - - nc, js := jsClient(t, s) - defer nc.Close() - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", LimitMarkerTTL: 100 * time.Millisecond}) - expectErr(t, err, jetstream.ErrInvalidLimitMarkerTTL) - - // now create a kv without LimitMarkerTTL - kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS"}) - expectOk(t, err) - - _, err = kv.Create(ctx, "age", []byte("22"), jetstream.KeyTTL(time.Second)) - expectErr(t, err, jetstream.ErrLimitMarkersNotEnabled) - - err = kv.Purge(ctx, "age", jetstream.PurgeTTL(time.Second)) - expectErr(t, err, jetstream.ErrLimitMarkersNotEnabled) - - // update kv to enable LimitMarkerTTL - kv, err = js.UpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", LimitMarkerTTL: time.Second}) - expectOk(t, err) - - // create a value with a TTL - _, err = kv.Create(ctx, "age", []byte("22"), jetstream.KeyTTL(time.Second)) - expectOk(t, err) - }) }