mirror of
https://github.com/1f349/dendrite.git
synced 2024-11-24 20:51:42 +00:00
Set max age for roomserver input stream to avoid excessive interior deletes (#3145)
If old messages build up in the input stream and do not get processed successfully, this can create a significant drift between the stream first sequence and the consumer ack floors, which results in a slow and expensive start-up when interest-based retention is in use. If a message is sat in the stream for 24 hours, it's probably not going to get processed successfully, so let NATS drop them instead. Dendrite can reconcile by fetching missing events later if it needs to. --------- Co-authored-by: Neil Alexander <neilalexander@users.noreply.github.com>
This commit is contained in:
parent
c08c7405db
commit
e93bdd56fd
@ -87,6 +87,7 @@ func (s *NATSInstance) Prepare(process *process.ProcessContext, cfg *config.JetS
|
|||||||
return js, nc
|
return js, nc
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// nolint:gocyclo
|
||||||
func setupNATS(process *process.ProcessContext, cfg *config.JetStream, nc *natsclient.Conn) (natsclient.JetStreamContext, *natsclient.Conn) {
|
func setupNATS(process *process.ProcessContext, cfg *config.JetStream, nc *natsclient.Conn) (natsclient.JetStreamContext, *natsclient.Conn) {
|
||||||
if nc == nil {
|
if nc == nil {
|
||||||
var err error
|
var err error
|
||||||
@ -126,18 +127,34 @@ func setupNATS(process *process.ProcessContext, cfg *config.JetStream, nc *natsc
|
|||||||
subjects = []string{name, name + ".>"}
|
subjects = []string{name, name + ".>"}
|
||||||
}
|
}
|
||||||
if info != nil {
|
if info != nil {
|
||||||
|
// If the stream config doesn't match what we expect, try to update
|
||||||
|
// it. If that doesn't work then try to blow it away and we'll then
|
||||||
|
// recreate it in the next section.
|
||||||
|
// Each specific option that we set must be checked by hand, as if
|
||||||
|
// you DeepEqual the whole config struct, it will always show that
|
||||||
|
// there's a difference because the NATS Server will return defaults
|
||||||
|
// in the stream info.
|
||||||
switch {
|
switch {
|
||||||
case !reflect.DeepEqual(info.Config.Subjects, subjects):
|
case !reflect.DeepEqual(info.Config.Subjects, subjects):
|
||||||
fallthrough
|
fallthrough
|
||||||
case info.Config.Retention != stream.Retention:
|
case info.Config.Retention != stream.Retention:
|
||||||
fallthrough
|
fallthrough
|
||||||
case info.Config.Storage != stream.Storage:
|
case info.Config.Storage != stream.Storage:
|
||||||
|
fallthrough
|
||||||
|
case info.Config.MaxAge != stream.MaxAge:
|
||||||
|
// Try updating the stream first, as many things can be updated
|
||||||
|
// non-destructively.
|
||||||
|
if info, err = s.UpdateStream(stream); err != nil {
|
||||||
|
logrus.WithError(err).Warnf("Unable to update stream %q, recreating...", name)
|
||||||
|
// We failed to update the stream, this is a last attempt to get
|
||||||
|
// things working but may result in data loss.
|
||||||
if err = s.DeleteStream(name); err != nil {
|
if err = s.DeleteStream(name); err != nil {
|
||||||
logrus.WithError(err).Fatal("Unable to delete stream")
|
logrus.WithError(err).Fatalf("Unable to delete stream %q", name)
|
||||||
}
|
}
|
||||||
info = nil
|
info = nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
if info == nil {
|
if info == nil {
|
||||||
// If we're trying to keep everything in memory (e.g. unit tests)
|
// If we're trying to keep everything in memory (e.g. unit tests)
|
||||||
// then overwrite the storage policy.
|
// then overwrite the storage policy.
|
||||||
|
@ -48,6 +48,7 @@ var streams = []*nats.StreamConfig{
|
|||||||
Name: InputRoomEvent,
|
Name: InputRoomEvent,
|
||||||
Retention: nats.InterestPolicy,
|
Retention: nats.InterestPolicy,
|
||||||
Storage: nats.FileStorage,
|
Storage: nats.FileStorage,
|
||||||
|
MaxAge: time.Hour * 24,
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
Name: InputDeviceListUpdate,
|
Name: InputDeviceListUpdate,
|
||||||
|
Loading…
Reference in New Issue
Block a user