Add NATS JetStream support

[S7evinK]

This adds experimental support for NATS JetStream by implementing the required interfaces for sarama.

Currently uses my experimental NATS JetStream sarama implementation, seems like there are still a few issues in there, as e.g. receipts/read markers don't seem to always come through.

On a side note, I was able to join #dendrite-dev and still receive messages in other rooms.

Kafka (with 3 nodes, 1 zookeeper) is using about ~4G of memory.. NATS on the other side (also 3 nodes, no manager needed) - 120M. (As reported by Prometheus) Kafka/NATS are both idle.

Some screenshots from NATS:

Pull Request Checklist

• I have added any new tests that need to pass to sytest-whitelist as specified in docs/sytest.md
• Pull request includes a sign off

[kegsay]

Fantastic work on this, thanks @S7evinK for all of your contributions thus far! I'll try to block out some time to look at this, but it'll take a while to migrate fully to NATS. We still need to sketch out and profile whether the swap would be net positive for large deployments. The main drivers for wanting to swap away from Kafka/Naffka are:

• Unbounded log growth - yes Kafka has a retention period but it's become clearer as the project has developed that we primarily want tiny retention periods, particularly for embedded use cases.
• Inability to send ephemeral events (generic pubsub to allow us to request actions on other components and register callbacks). Being able to do this will open up more ways to prevent race conditions between components.

[S7evinK]

Even after some more testing and trying to fix things, still the same issue with receipts not coming through. Looks like, at least that's what i guess, that only m.fully_read make their way to Element.

After a few days in #dendrite:

# nats stream report
Obtaining Stream stats

+---------------------------------------------------------------------------------------------------------------+
|                                                 Stream Report                                                 |
+---------------------------------+---------+-----------+----------+---------+------+---------+-----------------+
| Stream                          | Storage | Consumers | Messages | Bytes   | Lost | Deleted | Replicas        |
+---------------------------------+---------+-----------+----------+---------+------+---------+-----------------+
| DendriteOutputSendToDeviceEvent | File    | 2         | 0        | 0 B     | 0    | 0       | nats-cluster-0* |
| DendriteOutputTypingEvent       | File    | 2         | 0        | 0 B     | 0    | 0       | nats-cluster-0* |
| DendriteOutputClientData        | File    | 1         | 57       | 9.7 KiB | 0    | 0       | nats-cluster-0* |
| DendriteOutputReceiptEvent      | File    | 2         | 535      | 164 KiB | 0    | 0       | nats-cluster-2* |
| DendriteOutputKeyChangeEvent    | File    | 2         | 798      | 711 KiB | 0    | 0       | nats-cluster-2* |
| DendriteOutputRoomEvent         | File    | 2         | 1,537    | 4.0 MiB | 0    | 0       | nats-cluster-1* |
+---------------------------------+---------+-----------+----------+---------+------+---------+-----------------+

[kegsay]

Two main complaints:

• Phony changes are an unwelcome addition to this already enormous PR.
• Scattered Ack() calls throughout consumer code makes me extremely nervous that we will forget to ACK in places and I don't know what happens then? I guess it will get stuck on that event which is obviously not ideal.

[kegsay]

Why do we call InProgress twice for the same message?

[neilalexander]

This was because of the fact that the inbox might not process the task until later, so we reset the delivery timer when we start working on the event as well. As above though in the other comment, I don't think we actually need these at all, so I will remove them.

[kegsay]

Why do we do this here but in other cases we just Ack() or Nack()?

[neilalexander]

This was to do with trying to queue things up using request-reply, rather than putting things straight onto the inboxes themselves for synchronous requests. I never finished it and decided it would be better to do that in a separate PR later, but the Term here is pretty much equivalent here to an Ack.

[kegsay]

LGTM - Let's do it!

[r3k2]

Awesome! how do we change our current server/setup to test this?

[neilalexander]

Take a look at the jetstream section in the version 2 sample config file and make sure you have a store_dir that is read-writable by the Dendrite process, and you should be able to just upgrade to it with any luck. Let me know if you have any problems.