diff --git a/go_test.mod b/go_test.mod index 441162ddb..14bed41b0 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.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 ) diff --git a/go_test.sum b/go_test.sum index 2abf354a6..a45f05fb9 100644 --- a/go_test.sum +++ b/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= diff --git a/jetstream/jetstream_options.go b/jetstream/jetstream_options.go index 2712fc83b..6a6a2c909 100644 --- a/jetstream/jetstream_options.go +++ b/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. diff --git a/jetstream/message.go b/jetstream/message.go index 720a4a577..c34ffde0e 100644 --- a/jetstream/message.go +++ b/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" diff --git a/jetstream/publish.go b/jetstream/publish.go index 4c6af5942..b3e2993c9 100644 --- a/jetstream/publish.go +++ b/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 != "" { diff --git a/jetstream/stream_config.go b/jetstream/stream_config.go index 304203bc5..21083ca8a 100644 --- a/jetstream/stream_config.go +++ b/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 diff --git a/jetstream/test/publish_test.go b/jetstream/test/publish_test.go index f79ad19fb..c1fb3d83b 100644 --- a/jetstream/test/publish_test.go +++ b/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 { diff --git a/js.go b/js.go index 50c97cccf..bf890ee76 100644 --- a/js.go +++ b/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 diff --git a/jsm.go b/jsm.go index 2ae19c7a3..18ee255f2 100644 --- a/jsm.go +++ b/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. diff --git a/test/js_test.go b/test/js_test.go index 9db253344..a872d668f 100644 --- a/test/js_test.go +++ b/test/js_test.go @@ -430,6 +430,102 @@ func TestJetStreamPublish(t *testing.T) { } } +func TestPublishWithTTL(t *testing.T) { + srv := RunBasicJetStreamServer() + defer shutdownJSServerAndRemoveStorage(t, srv) + nc, js := jsClient(t, srv) + defer nc.Close() + + _, err := js.AddStream(&nats.StreamConfig{ + Name: "foo", Subjects: []string{"FOO.*"}, MaxMsgSize: 64, AllowMsgTTL: true}) + if err != nil { + t.Fatalf("Unexpected error: %v", err) + } + + ack, err := js.Publish("FOO.1", []byte("msg"), nats.MsgTTL(1*time.Second)) + if err != nil { + t.Fatalf("Unexpected error: %v", err) + } + gotMsg, err := js.GetMsg("foo", 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 = js.GetMsg("foo", ack.Sequence) + if !errors.Is(err, nats.ErrMsgNotFound) { + t.Fatalf("Expected not found error; got: %v", err) + } +} + +func TestMsgDeleteMarkerMaxAge(t *testing.T) { + srv := RunBasicJetStreamServer() + defer shutdownJSServerAndRemoveStorage(t, srv) + nc, js := jsClient(t, srv) + defer nc.Close() + + _, err := js.AddStream(&nats.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("FOO.1", []byte("msg1")) + if err != nil { + t.Fatalf("Unexpected error: %v", err) + } + time.Sleep(1500 * time.Millisecond) + gotMsg, err := js.GetLastMsg("foo", "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, js := jsClient(t, srv) + defer nc.Close() + + _, err := js.AddStream(&nats.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"), nats.MsgTTL(1*time.Second)) + if err != nil { + t.Fatalf("Unexpected error: %v", err) + } + var ack *nats.PubAck + select { + case ack = <-paf.Ok(): + case <-time.After(5 * time.Second): + t.Fatalf("Did not receive ack") + } + + gotMsg, err := js.GetMsg("foo", 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 = js.GetMsg("foo", ack.Sequence) + if !errors.Is(err, nats.ErrMsgNotFound) { + t.Fatalf("Expected not found error; got: %v", err) + } +} + func TestJetStreamSubscribe(t *testing.T) { s := RunBasicJetStreamServer() defer shutdownJSServerAndRemoveStorage(t, s)