[ADDED] Add per msg ttl

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.10.26
+	github.com/nats-io/nats-server/v2 v2.11.0-RC.5
 	github.com/nats-io/nkeys v0.4.10
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/goleak v1.3.0
@@ -15,9 +15,10 @@ 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
-	golang.org/x/crypto v0.34.0 // indirect
-	golang.org/x/sys v0.30.0 // indirect
-	golang.org/x/time v0.10.0 // indirect
+	golang.org/x/crypto v0.36.0 // indirect
+	golang.org/x/sys v0.31.0 // indirect
+	golang.org/x/time v0.11.0 // indirect
 )

go_test.sum

@@ -1,3 +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=
@@ -9,8 +11,11 @@ github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
-github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
+github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-tpm v0.9.3 h1:+yx0/anQuGzi+ssRqeD6WpXjW2L/V0dItUayO0i9sRc=
+github.com/google/go-tpm v0.9.3/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/minio/highwayhash v1.0.3 h1:kbnuUMoHYyVl7szWjSxJnxw11k2U709jqFPPmIUyD6Q=
@@ -19,8 +24,8 @@ 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.10.26 h1:2i3rAsn4x5/2eOt2NEmuI/iSb8zfHpIUI7yiaOWbo2c=
-github.com/nats-io/nats-server/v2 v2.10.26/go.mod h1:SGzoWGU8wUVnMr/HJhEMv4R8U4f7hF4zDygmRxpNsvg=
+github.com/nats-io/nats-server/v2 v2.11.0-RC.5 h1:/BAzHdcweWTTZpfQdYrOoBOQKDsCyfSLb0HccezAyYc=
+github.com/nats-io/nats-server/v2 v2.11.0-RC.5/go.mod h1:leXySghbdtXSUmWem8K9McnJ6xbJOb0t9+NQ5HTRZjI=
 github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
 github.com/nats-io/nkeys v0.4.10 h1:glmRrpCmYLHByYcePvnTBEAwawwapjCPMjy2huw20wc=
 github.com/nats-io/nkeys v0.4.10/go.mod h1:OjRrnIKnWBFl+s4YK5ChQfvHP2fxqZexrKJoVVyWB3U=
@@ -34,20 +39,19 @@ 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.34.0 h1:+/C6tk6rf/+t5DhUketUbD1aNGqiSX3j15Z6xuIDlBA=
-golang.org/x/crypto v0.34.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ=
+golang.org/x/crypto v0.36.0 h1:AnAEvhDddvBdpY+uR+MyHmuZzzNqXSe/GvuDeob5L34=
+golang.org/x/crypto v0.36.0/go.mod h1:Y4J0ReaxCR1IMaabaSMugxJES1EpwhBHhv2bDHklZvc=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
-golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
+golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
 golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
-golang.org/x/time v0.10.0 h1:3usCWA8tQn0L8+hFJQNgzpWbd89begxN66o1Ojdn5L4=
-golang.org/x/time v0.10.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+golang.org/x/time v0.11.0 h1:/bpjEDfN9tkoN/ryeYHnv5hcMlc8ncjMcM4XBk5NWV0=
+golang.org/x/time v0.11.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=

jetstream/jetstream_options.go

@@ -416,6 +416,15 @@ func WithMsgID(id string) PublishOpt {
 	}
 }
 
+// WithMsgTTL sets per msg TTL.
+// Requires [StreamConfig.AllowMsgTTL] to be enabled.
+func WithMsgTTL(dur time.Duration) PublishOpt {
+	return func(opts *pubOpts) error {
+		opts.ttl = dur
+		return nil
+	}
+}
+
 // WithExpectStream sets the expected stream the message should be published to.
 // If the message is published to a different stream server will reject the
 // message and publish will fail.

jetstream/message.go

@@ -190,6 +190,10 @@ const (
 	// option.
 	ExpectedLastMsgIDHeader = "Nats-Expected-Last-Msg-Id"
 
+	// MsgTTLHeader is used to specify the TTL for a specific message. This will
+	// override the default TTL for the stream.
+	MsgTTLHeader = "Nats-TTL"
+
 	// 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"

jetstream/publish.go

@@ -42,10 +42,11 @@ type (
 
 	pubOpts struct {
 		id             string
-		lastMsgID      string  // Expected last msgId
-		stream         string  // Expected stream name
-		lastSeq        *uint64 // Expected last sequence
-		lastSubjectSeq *uint64 // Expected last sequence per subject
+		lastMsgID      string        // Expected last msgId
+		stream         string        // Expected stream name
+		lastSeq        *uint64       // Expected last sequence
+		lastSubjectSeq *uint64       // Expected last sequence per subject
+		ttl            time.Duration // Message TTL
 
 		// Publish retries for NoResponders err.
 		retryWait     time.Duration // Retry wait between attempts
@@ -191,6 +192,9 @@ func (js *jetStream) PublishMsg(ctx context.Context, m *nats.Msg, opts ...Publis
 	if o.lastSubjectSeq != nil {
 		m.Header.Set(ExpectedLastSubjSeqHeader, strconv.FormatUint(*o.lastSubjectSeq, 10))
 	}
+	if o.ttl > 0 {
+		m.Header.Set(MsgTTLHeader, o.ttl.String())
+	}
 
 	var resp *nats.Msg
 	var err error
@@ -274,6 +278,9 @@ func (js *jetStream) PublishMsgAsync(m *nats.Msg, opts ...PublishOpt) (PubAckFut
 	if o.lastSubjectSeq != nil {
 		m.Header.Set(ExpectedLastSubjSeqHeader, strconv.FormatUint(*o.lastSubjectSeq, 10))
 	}
+	if o.ttl > 0 {
+		m.Header.Set(MsgTTLHeader, o.ttl.String())
+	}
 
 	paf := o.pafRetry
 	if paf == nil && m.Reply != "" {

jetstream/stream_config.go

@@ -196,6 +196,14 @@ type (
 		// Template identifies the template that manages the Stream.
 		// Deprecated: This feature is no longer supported.
 		Template string `json:"template_owner,omitempty"`
+
+		// AllowMsgTTL allows header initiated per-message TTLs.
+		// This feature requires nats-server v2.11.0 or later.
+		AllowMsgTTL bool `json:"allow_msg_ttl"`
+
+		// Enables and sets a duration for adding server markers for delete, purge and max age limits.
+		// This feature requires nats-server v2.11.0 or later.
+		SubjectDeleteMarkerTTL time.Duration `json:"subject_delete_marker_ttl,omitempty"`
 	}
 
 	// StreamSourceInfo shows information about an upstream stream

jetstream/test/publish_test.go

@@ -581,6 +581,132 @@ func TestPublishMsg(t *testing.T) {
 	}
 }
 
+func TestPublishWithTTL(t *testing.T) {
+	srv := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, srv)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer nc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	stream, err := js.CreateStream(ctx, jetstream.StreamConfig{
+		Name: "foo", Subjects: []string{"FOO.*"}, MaxMsgSize: 64, AllowMsgTTL: true})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	ack, err := js.Publish(ctx, "FOO.1", []byte("msg"), jetstream.WithMsgTTL(1*time.Second))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	gotMsg, err := stream.GetMsg(ctx, ack.Sequence)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if ttl := gotMsg.Header.Get("Nats-TTL"); ttl != "1s" {
+		t.Fatalf("Expected message to have TTL header set to 1s; got: %s", ttl)
+	}
+	time.Sleep(1500 * time.Millisecond)
+	_, err = stream.GetMsg(ctx, ack.Sequence)
+	if !errors.Is(err, jetstream.ErrMsgNotFound) {
+		t.Fatalf("Expected not found error; got: %v", err)
+	}
+}
+
+func TestMsgDeleteMarkerMaxAge(t *testing.T) {
+	srv := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, srv)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer nc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	stream, err := js.CreateStream(ctx, jetstream.StreamConfig{
+		Name: "foo", Subjects: []string{"FOO.*"}, AllowMsgTTL: true, SubjectDeleteMarkerTTL: 50 * time.Second, MaxAge: 1 * time.Second})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	_, err = js.Publish(ctx, "FOO.1", []byte("msg1"))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	time.Sleep(1500 * time.Millisecond)
+	gotMsg, err := stream.GetLastMsgForSubject(ctx, "FOO.1")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if ttlMarker := gotMsg.Header.Get("Nats-Marker-Reason"); ttlMarker != "MaxAge" {
+		t.Fatalf("Expected message to have Marker-Reason header set to MaxAge; got: %s", ttlMarker)
+	}
+	if ttl := gotMsg.Header.Get("Nats-TTL"); ttl != "50s" {
+		t.Fatalf("Expected message to have Nats-TTL header set to 50s; got: %s", ttl)
+	}
+}
+
+func TestPublishAsyncWithTTL(t *testing.T) {
+	srv := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, srv)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer nc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	stream, err := js.CreateStream(ctx, jetstream.StreamConfig{
+		Name: "foo", Subjects: []string{"FOO.*"}, MaxMsgSize: 64, AllowMsgTTL: true})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	paf, err := js.PublishAsync("FOO.1", []byte("msg"), jetstream.WithMsgTTL(1*time.Second))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	var ack *jetstream.PubAck
+	select {
+	case ack = <-paf.Ok():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive ack")
+	}
+
+	gotMsg, err := stream.GetMsg(ctx, ack.Sequence)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if ttl := gotMsg.Header.Get("Nats-TTL"); ttl != "1s" {
+		t.Fatalf("Expected message to have TTL header set to 1s; got: %s", ttl)
+	}
+	time.Sleep(1500 * time.Millisecond)
+	_, err = stream.GetMsg(ctx, ack.Sequence)
+	if !errors.Is(err, jetstream.ErrMsgNotFound) {
+		t.Fatalf("Expected not found error; got: %v", err)
+	}
+}
+
 func TestPublish(t *testing.T) {
 	// Only very basic test cases, as most use cases are tested in TestPublishMsg
 	tests := []struct {

js.go

@@ -466,13 +466,14 @@ func (opt pubOptFn) configurePublish(opts *pubOpts) error {
 }
 
 type pubOpts struct {
-	ctx context.Context
-	ttl time.Duration
-	id  string
-	lid string  // Expected last msgId
-	str string  // Expected stream name
-	seq *uint64 // Expected last sequence
-	lss *uint64 // Expected last sequence per subject
+	ctx    context.Context
+	ttl    time.Duration
+	id     string
+	lid    string        // Expected last msgId
+	str    string        // Expected stream name
+	seq    *uint64       // Expected last sequence
+	lss    *uint64       // Expected last sequence per subject
+	msgTTL time.Duration // Message TTL
 
 	// Publish retries for NoResponders err.
 	rwait time.Duration // Retry wait between attempts
@@ -507,6 +508,7 @@ const (
 	ExpectedLastSubjSeqHdr = "Nats-Expected-Last-Subject-Sequence"
 	ExpectedLastMsgIdHdr   = "Nats-Expected-Last-Msg-Id"
 	MsgRollup              = "Nats-Rollup"
+	MsgTTLHdr              = "Nats-TTL"
 )
 
 // Headers for republished messages and direct gets.
@@ -566,6 +568,9 @@ func (js *js) PublishMsg(m *Msg, opts ...PubOpt) (*PubAck, error) {
 	if o.lss != nil {
 		m.Header.Set(ExpectedLastSubjSeqHdr, strconv.FormatUint(*o.lss, 10))
 	}
+	if o.msgTTL > 0 {
+		m.Header.Set(MsgTTLHdr, o.msgTTL.String())
+	}
 
 	var resp *Msg
 	var err error
@@ -1024,6 +1029,9 @@ func (js *js) PublishMsgAsync(m *Msg, opts ...PubOpt) (PubAckFuture, error) {
 	if o.lss != nil {
 		m.Header.Set(ExpectedLastSubjSeqHdr, strconv.FormatUint(*o.lss, 10))
 	}
+	if o.msgTTL > 0 {
+		m.Header.Set(MsgTTLHdr, o.msgTTL.String())
+	}
 
 	// Reply
 	paf := o.pafRetry
@@ -1151,6 +1159,15 @@ func StallWait(ttl time.Duration) PubOpt {
 	})
 }
 
+// MsgTTL sets per msg TTL.
+// Requires [StreamConfig.AllowMsgTTL] to be enabled.
+func MsgTTL(dur time.Duration) PubOpt {
+	return pubOptFn(func(opts *pubOpts) error {
+		opts.msgTTL = dur
+		return nil
+	})
+}
+
 type ackOpts struct {
 	ttl      time.Duration
 	ctx      context.Context

jsm.go

@@ -243,6 +243,14 @@ type StreamConfig struct {
 	// Template identifies the template that manages the Stream. Deprecated:
 	// This feature is no longer supported.
 	Template string `json:"template_owner,omitempty"`
+
+	// AllowMsgTTL allows header initiated per-message TTLs.
+	// This feature requires nats-server v2.11.0 or later.
+	AllowMsgTTL bool `json:"allow_msg_ttl"`
+
+	// Enables and sets a duration for adding server markers for delete, purge and max age limits.
+	// This feature requires nats-server v2.11.0 or later.
+	SubjectDeleteMarkerTTL time.Duration `json:"subject_delete_marker_ttl,omitempty"`
 }
 
 // SubjectTransformConfig is for applying a subject transform (to matching messages) before doing anything else when a new message is received.