Initial support for per-message TTLs #6272
Conversation
There was a problem hiding this comment.
LGTM
Found and fixed two bugs; if there was no index.db the TTLs would not be recovered (with or without thw.db), and if a single message needed to be recovered (when no thw.db) it would not due to an off-by-one.
Still work to be done in future PRs, but think this one is good. 👍
server/filestore.go
Outdated
| fs.ttls.ExpireTasks(func(seq uint64, ts int64) { | ||
| fs.removeMsgViaLimits(seq) | ||
| }) | ||
| nextTTL = fs.ttls.GetNextExpiration(math.MaxInt64) |
There was a problem hiding this comment.
This should only pass max int if we are not doing max age, if max age then we should capture the next fireIn from the loop above and pass that here.
There was a problem hiding this comment.
Did we get this changed as discussed?
There was a problem hiding this comment.
Not yet, I'm still fiddling with this and want to write a proper test for it. Happy to do in either this PR or in a follow-up one.
server/stream.go
Outdated
| @@ -4996,17 +5011,20 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte, | |||
| } | |||
| } | |||
|
|
|||
| // Find the message TTL if any. | |||
| ttl := getMessageTTL(hdr) | |||
There was a problem hiding this comment.
Reject if ttl present and stream is not opted in.
There was a problem hiding this comment.
I don't think we can do this, otherwise we can run into a problem when we source/mirror messages with a TTL into another stream.
There was a problem hiding this comment.
We need special processing for streams and mirrors anyway that ignores (for ingest criteria) certain headers. This would be one added to the list.
There was a problem hiding this comment.
Will add that to the list of follow-up changes to make.
neilalexander
force-pushed
the
neil/ttl
branch
from
4b5b21e to
55537d2
Compare
server/filestore.go
Outdated
| fs.ttls.ExpireTasks(func(seq uint64, ts int64) { | ||
| fs.removeMsgViaLimits(seq) | ||
| }) | ||
| nextTTL = fs.ttls.GetNextExpiration(math.MaxInt64) |
There was a problem hiding this comment.
Did we get this changed as discussed?
server/filestore_test.go
Outdated
|
|
||
| func TestFileStoreMessageTTL(t *testing.T) { | ||
| fs, err := newFileStore( | ||
| FileStoreConfig{StoreDir: t.TempDir(), EnforceTTLs: true}, |
There was a problem hiding this comment.
Can we not inherit enforeTTLS from the stream config?
There was a problem hiding this comment.
In stream.go when creating the store for streams, we do indeed set it from the AllowMsgTTLs value from the stream config. In the filestore-specific tests, like here, we set it to true directly.
There was a problem hiding this comment.
Since we pass in a stream config in the tests we could do it there, or you feel we should keep the redundancy and have it in two places?
There was a problem hiding this comment.
Not attached to the duplication at all, will remove it and use the passed-in StreamConfig one.
| require_Equal(t, ss.LastSeq, 1) | ||
| require_Equal(t, ss.Msgs, 0) | ||
| }) | ||
| } |
There was a problem hiding this comment.
Need to add in test with max age but with msgs with never expire and make sure they do not get expired.
There was a problem hiding this comment.
At the moment this PR doesn't have "never expire" behaviour for MaxAge, that's one of the things I have queued up for a follow-up PR.
There was a problem hiding this comment.
ok maybe remove the header then from this PR. Confusing as just a placeholder.
| @@ -100,6 +100,10 @@ type StreamConfig struct { | |||
| // TODO(nat): Can/should we name these better? | |||
| ConsumerLimits StreamConsumerLimits `json:"consumer_limits"` | |||
|
|
|||
| // AllowMsgTTL allows header initiated per-message TTLs. If disabled, | |||
| // then the `NATS-TTL` header will be ignored. | |||
There was a problem hiding this comment.
Update comment if we reject these vs ignore.
There was a problem hiding this comment.
At the moment the comment is correct, as of right now the header is passed through and ignored. We probably want to figure out the sourcing/mirroring behaviour at the same time as changing this.
There was a problem hiding this comment.
I saw the thread between you and R.I. and Byron. From an app perspective if I set a header and get no feedback it is not honored that is not good IMO.
Yes we need to fixup the sourcing and mirroring to ignore certain headers. It does so already on some but we need a better solution here.
neilalexander
force-pushed
the
neil/ttl
branch
2 times, most recently
from
621a876 to
4d4187b
Compare
|
Have rebased the PR, I think it's best to deal with rejecting the TTL'd messages in a separate PR at the same time as stripping them from sourcing/mirroring. |
Signed-off-by: Neil Twigg <[email protected]>
… that when checking for next expiry Signed-off-by: Neil Twigg <[email protected]>
neilalexander
force-pushed
the
neil/ttl
branch
from
4d4187b to
6ea9fe0
Compare
|
Fixed merge conflicts. |
This is an incomplete but somewhat-working implementations of per-message TTLs.
Notes:
AllowMsgTTLto stream config;Nats-TTLmessage header as a parsable duration, or if not, second precision;index.dbmagic version so that we can add a new field for tracking how many TTL'd messages are in each message block (means thatindex.dbneeds to be rebuilt if downgrading);thw.dbto store the timed hash wheel state, and tries to rebuild it from a linear scan if it is missing or out-of-date.Future work:
Signed-off-by: Neil Twigg [email protected]