1f349/dendrite
4
0
Fork 0
mirror of https://github.com/1f349/dendrite.git synced 2024-11-22 11:41:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use pull consumers (#2140)

Browse Source 
* Pull consumers

* Pull consumers

* Only nuke consumers if they are push consumers

* Clean up old consumers

* Better error handling

* Update comments
This commit is contained in:
Neil Alexander 2022-02-02 13:32:48 +00:00 committed by GitHub
parent 2dee706f9e
commit c773b038bb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 521 additions and 459 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
appservice/consumers/roomserver.go
View File

@ -34,7 +34,7 @@ import (
type OutputRoomEventConsumer struct { type OutputRoomEventConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
asDB storage.Database asDB storage.Database
rsAPI api.RoomserverInternalAPI rsAPI api.RoomserverInternalAPI
@ -66,37 +66,37 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers // Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error { func (s *OutputRoomEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return jetstream.JetStreamConsumer(
return err s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
nats.DeliverAll(), nats.ManualAck(),
)
} }
// onMessage is called when the appservice component receives a new event from // onMessage is called when the appservice component receives a new event from
// the room server output log. // the room server output log.
func (s *OutputRoomEventConsumer) onMessage(msg *nats.Msg) { func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
jetstream.WithJetStreamMessage(msg, func(msg *nats.Msg) bool { // Parse out the event JSON
// Parse out the event JSON var output api.OutputEvent
var output api.OutputEvent if err := json.Unmarshal(msg.Data, &output); err != nil {
if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream
// If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("roomserver output log: message parse failure")
log.WithError(err).Errorf("roomserver output log: message parse failure")
return true
}
if output.Type != api.OutputTypeNewRoomEvent {
return true
}
events := []*gomatrixserverlib.HeaderedEvent{output.NewRoomEvent.Event}
events = append(events, output.NewRoomEvent.AddStateEvents...)
// Send event to any relevant application services
if err := s.filterRoomserverEvents(context.TODO(), events); err != nil {
log.WithError(err).Errorf("roomserver output log: filter error")
return true
}
return true return true
}) }
if output.Type != api.OutputTypeNewRoomEvent {
return true
}
events := []*gomatrixserverlib.HeaderedEvent{output.NewRoomEvent.Event}
events = append(events, output.NewRoomEvent.AddStateEvents...)
// Send event to any relevant application services
if err := s.filterRoomserverEvents(context.TODO(), events); err != nil {
log.WithError(err).Errorf("roomserver output log: filter error")
return true
}
return true
} }
// filterRoomserverEvents takes in events and decides whether any of them need // filterRoomserverEvents takes in events and decides whether any of them need

329
federationapi/consumers/eduserver.go
View File

@ -34,7 +34,7 @@ import (
type OutputEDUConsumer struct { type OutputEDUConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
db storage.Database db storage.Database
queues *queue.OutgoingQueues queues *queue.OutgoingQueues
ServerName gomatrixserverlib.ServerName ServerName gomatrixserverlib.ServerName
@ -66,13 +66,22 @@ func NewOutputEDUConsumer(
// Start consuming from EDU servers // Start consuming from EDU servers
func (t *OutputEDUConsumer) Start() error { func (t *OutputEDUConsumer) Start() error {
if _, err := t.jetstream.Subscribe(t.typingTopic, t.onTypingEvent, t.durable); err != nil { if err := jetstream.JetStreamConsumer(
t.ctx, t.jetstream, t.typingTopic, t.durable, t.onTypingEvent,
nats.DeliverAll(), nats.ManualAck(),
); err != nil {
return err return err
} }
if _, err := t.jetstream.Subscribe(t.sendToDeviceTopic, t.onSendToDeviceEvent, t.durable); err != nil { if err := jetstream.JetStreamConsumer(
t.ctx, t.jetstream, t.sendToDeviceTopic, t.durable, t.onSendToDeviceEvent,
nats.DeliverAll(), nats.ManualAck(),
); err != nil {
return err return err
} }
if _, err := t.jetstream.Subscribe(t.receiptTopic, t.onReceiptEvent, t.durable); err != nil { if err := jetstream.JetStreamConsumer(
t.ctx, t.jetstream, t.receiptTopic, t.durable, t.onReceiptEvent,
nats.DeliverAll(), nats.ManualAck(),
); err != nil {
return err return err
} }
return nil return nil
@ -80,175 +89,169 @@ func (t *OutputEDUConsumer) Start() error {
// onSendToDeviceEvent is called in response to a message received on the // onSendToDeviceEvent is called in response to a message received on the
// send-to-device events topic from the EDU server. // send-to-device events topic from the EDU server.
func (t *OutputEDUConsumer) onSendToDeviceEvent(msg *nats.Msg) { func (t *OutputEDUConsumer) onSendToDeviceEvent(ctx context.Context, msg *nats.Msg) bool {
// Extract the send-to-device event from msg. // Extract the send-to-device event from msg.
jetstream.WithJetStreamMessage(msg, func(msg *nats.Msg) bool { var ote api.OutputSendToDeviceEvent
var ote api.OutputSendToDeviceEvent if err := json.Unmarshal(msg.Data, &ote); err != nil {
if err := json.Unmarshal(msg.Data, &ote); err != nil { log.WithError(err).Errorf("eduserver output log: message parse failed (expected send-to-device)")
log.WithError(err).Errorf("eduserver output log: message parse failed (expected send-to-device)")
return true
}
// only send send-to-device events which originated from us
_, originServerName, err := gomatrixserverlib.SplitID('@', ote.Sender)
if err != nil {
log.WithError(err).WithField("user_id", ote.Sender).Error("Failed to extract domain from send-to-device sender")
return true
}
if originServerName != t.ServerName {
log.WithField("other_server", originServerName).Info("Suppressing send-to-device: originated elsewhere")
return true
}
_, destServerName, err := gomatrixserverlib.SplitID('@', ote.UserID)
if err != nil {
log.WithError(err).WithField("user_id", ote.UserID).Error("Failed to extract domain from send-to-device destination")
return true
}
// Pack the EDU and marshal it
edu := &gomatrixserverlib.EDU{
Type: gomatrixserverlib.MDirectToDevice,
Origin: string(t.ServerName),
}
tdm := gomatrixserverlib.ToDeviceMessage{
Sender: ote.Sender,
Type: ote.Type,
MessageID: util.RandomString(32),
Messages: map[string]map[string]json.RawMessage{
ote.UserID: {
ote.DeviceID: ote.Content,
},
},
}
if edu.Content, err = json.Marshal(tdm); err != nil {
log.WithError(err).Error("failed to marshal EDU JSON")
return true
}
log.Infof("Sending send-to-device message into %q destination queue", destServerName)
if err := t.queues.SendEDU(edu, t.ServerName, []gomatrixserverlib.ServerName{destServerName}); err != nil {
log.WithError(err).Error("failed to send EDU")
return false
}
return true return true
}) }
// only send send-to-device events which originated from us
_, originServerName, err := gomatrixserverlib.SplitID('@', ote.Sender)
if err != nil {
log.WithError(err).WithField("user_id", ote.Sender).Error("Failed to extract domain from send-to-device sender")
return true
}
if originServerName != t.ServerName {
log.WithField("other_server", originServerName).Info("Suppressing send-to-device: originated elsewhere")
return true
}
_, destServerName, err := gomatrixserverlib.SplitID('@', ote.UserID)
if err != nil {
log.WithError(err).WithField("user_id", ote.UserID).Error("Failed to extract domain from send-to-device destination")
return true
}
// Pack the EDU and marshal it
edu := &gomatrixserverlib.EDU{
Type: gomatrixserverlib.MDirectToDevice,
Origin: string(t.ServerName),
}
tdm := gomatrixserverlib.ToDeviceMessage{
Sender: ote.Sender,
Type: ote.Type,
MessageID: util.RandomString(32),
Messages: map[string]map[string]json.RawMessage{
ote.UserID: {
ote.DeviceID: ote.Content,
},
},
}
if edu.Content, err = json.Marshal(tdm); err != nil {
log.WithError(err).Error("failed to marshal EDU JSON")
return true
}
log.Infof("Sending send-to-device message into %q destination queue", destServerName)
if err := t.queues.SendEDU(edu, t.ServerName, []gomatrixserverlib.ServerName{destServerName}); err != nil {
log.WithError(err).Error("failed to send EDU")
return false
}
return true
} }
// onTypingEvent is called in response to a message received on the typing // onTypingEvent is called in response to a message received on the typing
// events topic from the EDU server. // events topic from the EDU server.
func (t *OutputEDUConsumer) onTypingEvent(msg *nats.Msg) { func (t *OutputEDUConsumer) onTypingEvent(ctx context.Context, msg *nats.Msg) bool {
jetstream.WithJetStreamMessage(msg, func(msg *nats.Msg) bool { // Extract the typing event from msg.
// Extract the typing event from msg. var ote api.OutputTypingEvent
var ote api.OutputTypingEvent if err := json.Unmarshal(msg.Data, &ote); err != nil {
if err := json.Unmarshal(msg.Data, &ote); err != nil { // Skip this msg but continue processing messages.
// Skip this msg but continue processing messages. log.WithError(err).Errorf("eduserver output log: message parse failed (expected typing)")
log.WithError(err).Errorf("eduserver output log: message parse failed (expected typing)") _ = msg.Ack()
_ = msg.Ack()
return true
}
// only send typing events which originated from us
_, typingServerName, err := gomatrixserverlib.SplitID('@', ote.Event.UserID)
if err != nil {
log.WithError(err).WithField("user_id", ote.Event.UserID).Error("Failed to extract domain from typing sender")
_ = msg.Ack()
return true
}
if typingServerName != t.ServerName {
return true
}
joined, err := t.db.GetJoinedHosts(t.ctx, ote.Event.RoomID)
if err != nil {
log.WithError(err).WithField("room_id", ote.Event.RoomID).Error("failed to get joined hosts for room")
return false
}
names := make([]gomatrixserverlib.ServerName, len(joined))
for i := range joined {
names[i] = joined[i].ServerName
}
edu := &gomatrixserverlib.EDU{Type: ote.Event.Type}
if edu.Content, err = json.Marshal(map[string]interface{}{
"room_id": ote.Event.RoomID,
"user_id": ote.Event.UserID,
"typing": ote.Event.Typing,
}); err != nil {
log.WithError(err).Error("failed to marshal EDU JSON")
return true
}
if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil {
log.WithError(err).Error("failed to send EDU")
return false
}
return true return true
}) }
// only send typing events which originated from us
_, typingServerName, err := gomatrixserverlib.SplitID('@', ote.Event.UserID)
if err != nil {
log.WithError(err).WithField("user_id", ote.Event.UserID).Error("Failed to extract domain from typing sender")
_ = msg.Ack()
return true
}
if typingServerName != t.ServerName {
return true
}
joined, err := t.db.GetJoinedHosts(ctx, ote.Event.RoomID)
if err != nil {
log.WithError(err).WithField("room_id", ote.Event.RoomID).Error("failed to get joined hosts for room")
return false
}
names := make([]gomatrixserverlib.ServerName, len(joined))
for i := range joined {
names[i] = joined[i].ServerName
}
edu := &gomatrixserverlib.EDU{Type: ote.Event.Type}
if edu.Content, err = json.Marshal(map[string]interface{}{
"room_id": ote.Event.RoomID,
"user_id": ote.Event.UserID,
"typing": ote.Event.Typing,
}); err != nil {
log.WithError(err).Error("failed to marshal EDU JSON")
return true
}
if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil {
log.WithError(err).Error("failed to send EDU")
return false
}
return true
} }
// onReceiptEvent is called in response to a message received on the receipt // onReceiptEvent is called in response to a message received on the receipt
// events topic from the EDU server. // events topic from the EDU server.
func (t *OutputEDUConsumer) onReceiptEvent(msg *nats.Msg) { func (t *OutputEDUConsumer) onReceiptEvent(ctx context.Context, msg *nats.Msg) bool {
jetstream.WithJetStreamMessage(msg, func(msg *nats.Msg) bool { // Extract the typing event from msg.
// Extract the typing event from msg. var receipt api.OutputReceiptEvent
var receipt api.OutputReceiptEvent if err := json.Unmarshal(msg.Data, &receipt); err != nil {
if err := json.Unmarshal(msg.Data, &receipt); err != nil { // Skip this msg but continue processing messages.
// Skip this msg but continue processing messages. log.WithError(err).Errorf("eduserver output log: message parse failed (expected receipt)")
log.WithError(err).Errorf("eduserver output log: message parse failed (expected receipt)")
return true
}
// only send receipt events which originated from us
_, receiptServerName, err := gomatrixserverlib.SplitID('@', receipt.UserID)
if err != nil {
log.WithError(err).WithField("user_id", receipt.UserID).Error("failed to extract domain from receipt sender")
return true
}
if receiptServerName != t.ServerName {
return true
}
joined, err := t.db.GetJoinedHosts(t.ctx, receipt.RoomID)
if err != nil {
log.WithError(err).WithField("room_id", receipt.RoomID).Error("failed to get joined hosts for room")
return false
}
names := make([]gomatrixserverlib.ServerName, len(joined))
for i := range joined {
names[i] = joined[i].ServerName
}
content := map[string]api.FederationReceiptMRead{}
content[receipt.RoomID] = api.FederationReceiptMRead{
User: map[string]api.FederationReceiptData{
receipt.UserID: {
Data: api.ReceiptTS{
TS: receipt.Timestamp,
},
EventIDs: []string{receipt.EventID},
},
},
}
edu := &gomatrixserverlib.EDU{
Type: gomatrixserverlib.MReceipt,
Origin: string(t.ServerName),
}
if edu.Content, err = json.Marshal(content); err != nil {
log.WithError(err).Error("failed to marshal EDU JSON")
return true
}
if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil {
log.WithError(err).Error("failed to send EDU")
return false
}
return true return true
}) }
// only send receipt events which originated from us
_, receiptServerName, err := gomatrixserverlib.SplitID('@', receipt.UserID)
if err != nil {
log.WithError(err).WithField("user_id", receipt.UserID).Error("failed to extract domain from receipt sender")
return true
}
if receiptServerName != t.ServerName {
return true
}
joined, err := t.db.GetJoinedHosts(ctx, receipt.RoomID)
if err != nil {
log.WithError(err).WithField("room_id", receipt.RoomID).Error("failed to get joined hosts for room")
return false
}
names := make([]gomatrixserverlib.ServerName, len(joined))
for i := range joined {
names[i] = joined[i].ServerName
}
content := map[string]api.FederationReceiptMRead{}
content[receipt.RoomID] = api.FederationReceiptMRead{
User: map[string]api.FederationReceiptData{
receipt.UserID: {
Data: api.ReceiptTS{
TS: receipt.Timestamp,
},
EventIDs: []string{receipt.EventID},
},
},
}
edu := &gomatrixserverlib.EDU{
Type: gomatrixserverlib.MReceipt,
Origin: string(t.ServerName),
}
if edu.Content, err = json.Marshal(content); err != nil {
log.WithError(err).Error("failed to marshal EDU JSON")
return true
}
if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil {
log.WithError(err).Error("failed to send EDU")
return false
}
return true
} }

108
federationapi/consumers/roomserver.go
View File

@ -37,7 +37,7 @@ type OutputRoomEventConsumer struct {
cfg *config.FederationAPI cfg *config.FederationAPI
rsAPI api.RoomserverInternalAPI rsAPI api.RoomserverInternalAPI
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
db storage.Database db storage.Database
queues *queue.OutgoingQueues queues *queue.OutgoingQueues
topic string topic string
@ -66,74 +66,70 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers // Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error { func (s *OutputRoomEventConsumer) Start() error {
_, err := s.jetstream.Subscribe( return jetstream.JetStreamConsumer(
s.topic, s.onMessage, s.durable, s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
nats.DeliverAll(), nats.DeliverAll(), nats.ManualAck(),
nats.ManualAck(),
) )
return err
} }
// onMessage is called when the federation server receives a new event from the room server output log. // onMessage is called when the federation server receives a new event from the room server output log.
// It is unsafe to call this with messages for the same room in multiple gorountines // It is unsafe to call this with messages for the same room in multiple gorountines
// because updates it will likely fail with a types.EventIDMismatchError when it // because updates it will likely fail with a types.EventIDMismatchError when it
// realises that it cannot update the room state using the deltas. // realises that it cannot update the room state using the deltas.
func (s *OutputRoomEventConsumer) onMessage(msg *nats.Msg) { func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
jetstream.WithJetStreamMessage(msg, func(msg *nats.Msg) bool { // Parse out the event JSON
// Parse out the event JSON var output api.OutputEvent
var output api.OutputEvent if err := json.Unmarshal(msg.Data, &output); err != nil {
if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream
// If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("roomserver output log: message parse failure")
log.WithError(err).Errorf("roomserver output log: message parse failure") return true
return true }
}
switch output.Type { switch output.Type {
case api.OutputTypeNewRoomEvent: case api.OutputTypeNewRoomEvent:
ev := output.NewRoomEvent.Event ev := output.NewRoomEvent.Event
if output.NewRoomEvent.RewritesState { if output.NewRoomEvent.RewritesState {
if err := s.db.PurgeRoomState(s.ctx, ev.RoomID()); err != nil { if err := s.db.PurgeRoomState(s.ctx, ev.RoomID()); err != nil {
log.WithError(err).Errorf("roomserver output log: purge room state failure") log.WithError(err).Errorf("roomserver output log: purge room state failure")
return false
}
}
if err := s.processMessage(*output.NewRoomEvent); err != nil {
switch err.(type) {
case *queue.ErrorFederationDisabled:
log.WithField("error", output.Type).Info(
err.Error(),
)
default:
// panic rather than continue with an inconsistent database
log.WithFields(log.Fields{
"event_id": ev.EventID(),
"event": string(ev.JSON()),
"add": output.NewRoomEvent.AddsStateEventIDs,
"del": output.NewRoomEvent.RemovesStateEventIDs,
log.ErrorKey: err,
}).Panicf("roomserver output log: write room event failure")
}
}
case api.OutputTypeNewInboundPeek:
if err := s.processInboundPeek(*output.NewInboundPeek); err != nil {
log.WithFields(log.Fields{
"event": output.NewInboundPeek,
log.ErrorKey: err,
}).Panicf("roomserver output log: remote peek event failure")
return false return false
} }
default:
log.WithField("type", output.Type).Debug(
"roomserver output log: ignoring unknown output type",
)
} }
return true if err := s.processMessage(*output.NewRoomEvent); err != nil {
}) switch err.(type) {
case *queue.ErrorFederationDisabled:
log.WithField("error", output.Type).Info(
err.Error(),
)
default:
// panic rather than continue with an inconsistent database
log.WithFields(log.Fields{
"event_id": ev.EventID(),
"event": string(ev.JSON()),
"add": output.NewRoomEvent.AddsStateEventIDs,
"del": output.NewRoomEvent.RemovesStateEventIDs,
log.ErrorKey: err,
}).Panicf("roomserver output log: write room event failure")
}
}
case api.OutputTypeNewInboundPeek:
if err := s.processInboundPeek(*output.NewInboundPeek); err != nil {
log.WithFields(log.Fields{
"event": output.NewInboundPeek,
log.ErrorKey: err,
}).Panicf("roomserver output log: remote peek event failure")
return false
}
default:
log.WithField("type", output.Type).Debug(
"roomserver output log: ignoring unknown output type",
)
}
return true
} }
// processInboundPeek starts tracking a new federated inbound peek (replacing the existing one if any) // processInboundPeek starts tracking a new federated inbound peek (replacing the existing one if any)

4
roomserver/internal/api.go
View File

@ -41,7 +41,7 @@ type RoomserverInternalAPI struct {
fsAPI fsAPI.FederationInternalAPI fsAPI fsAPI.FederationInternalAPI
asAPI asAPI.AppServiceQueryAPI asAPI asAPI.AppServiceQueryAPI
JetStream nats.JetStreamContext JetStream nats.JetStreamContext
Durable nats.SubOpt Durable string
InputRoomEventTopic string // JetStream topic for new input room events InputRoomEventTopic string // JetStream topic for new input room events
OutputRoomEventTopic string // JetStream topic for new output room events OutputRoomEventTopic string // JetStream topic for new output room events
PerspectiveServerNames []gomatrixserverlib.ServerName PerspectiveServerNames []gomatrixserverlib.ServerName
@ -87,7 +87,7 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.FederationInternalA
InputRoomEventTopic: r.InputRoomEventTopic, InputRoomEventTopic: r.InputRoomEventTopic,
OutputRoomEventTopic: r.OutputRoomEventTopic, OutputRoomEventTopic: r.OutputRoomEventTopic,
JetStream: r.JetStream, JetStream: r.JetStream,
Durable: r.Durable, Durable: nats.Durable(r.Durable),
ServerName: r.Cfg.Matrix.ServerName, ServerName: r.Cfg.Matrix.ServerName,
FSAPI: fsAPI, FSAPI: fsAPI,
KeyRing: keyRing, KeyRing: keyRing,

6
setup/config/config_jetstream.go
View File

@ -2,8 +2,6 @@ package config
import ( import (
"fmt" "fmt"
"github.com/nats-io/nats.go"
) )
type JetStream struct { type JetStream struct {
@ -25,8 +23,8 @@ func (c *JetStream) TopicFor(name string) string {
return fmt.Sprintf("%s%s", c.TopicPrefix, name) return fmt.Sprintf("%s%s", c.TopicPrefix, name)
} }
func (c *JetStream) Durable(name string) nats.SubOpt { func (c *JetStream) Durable(name string) string {
return nats.Durable(c.TopicFor(name)) return c.TopicFor(name)
} }
func (c *JetStream) Defaults(generate bool) { func (c *JetStream) Defaults(generate bool) {

83
setup/jetstream/helpers.go
View File

@ -1,12 +1,81 @@
package jetstream package jetstream
import "github.com/nats-io/nats.go" import (
"context"
"fmt"
func WithJetStreamMessage(msg *nats.Msg, f func(msg *nats.Msg) bool) { "github.com/nats-io/nats.go"
_ = msg.InProgress() "github.com/sirupsen/logrus"
if f(msg) { )
_ = msg.Ack()
} else { func JetStreamConsumer(
_ = msg.Nak() ctx context.Context, js nats.JetStreamContext, subj, durable string,
f func(ctx context.Context, msg *nats.Msg) bool,
opts ...nats.SubOpt,
) error {
defer func() {
// If there are existing consumers from before they were pull
// consumers, we need to clean up the old push consumers. However,
// in order to not affect the interest-based policies, we need to
// do this *after* creating the new pull consumers, which have
// "Pull" suffixed to their name.
if _, err := js.ConsumerInfo(subj, durable); err == nil {
if err := js.DeleteConsumer(subj, durable); err != nil {
logrus.WithContext(ctx).Warnf("Failed to clean up old consumer %q", durable)
}
}
}()
name := durable + "Pull"
sub, err := js.PullSubscribe(subj, name, opts...)
if err != nil {
return fmt.Errorf("nats.SubscribeSync: %w", err)
} }
go func() {
for {
// The context behaviour here is surprising — we supply a context
// so that we can interrupt the fetch if we want, but NATS will still
// enforce its own deadline (roughly 5 seconds by default). Therefore
// it is our responsibility to check whether our context expired or
// not when a context error is returned. Footguns. Footguns everywhere.
msgs, err := sub.Fetch(1, nats.Context(ctx))
if err != nil {
if err == context.Canceled || err == context.DeadlineExceeded {
// Work out whether it was the JetStream context that expired
// or whether it was our supplied context.
select {
case <-ctx.Done():
// The supplied context expired, so we want to stop the
// consumer altogether.
return
default:
// The JetStream context expired, so the fetch probably
// just timed out and we should try again.
continue
}
} else {
// Something else went wrong, so we'll panic.
logrus.WithContext(ctx).WithField("subject", subj).Fatal(err)
}
}
if len(msgs) < 1 {
continue
}
msg := msgs[0]
if err = msg.InProgress(); err != nil {
logrus.WithContext(ctx).WithField("subject", subj).Warn(fmt.Errorf("msg.InProgress: %w", err))
continue
}
if f(ctx, msg) {
if err = msg.Ack(); err != nil {
logrus.WithContext(ctx).WithField("subject", subj).Warn(fmt.Errorf("msg.Ack: %w", err))
}
} else {
if err = msg.Nak(); err != nil {
logrus.WithContext(ctx).WithField("subject", subj).Warn(fmt.Errorf("msg.Nak: %w", err))
}
}
}
}()
return nil
} }

72
syncapi/consumers/clientapi.go
View File

@ -34,7 +34,7 @@ import (
type OutputClientDataConsumer struct { type OutputClientDataConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
db storage.Database db storage.Database
stream types.StreamProvider stream types.StreamProvider
@ -63,45 +63,45 @@ func NewOutputClientDataConsumer(
// Start consuming from room servers // Start consuming from room servers
func (s *OutputClientDataConsumer) Start() error { func (s *OutputClientDataConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return jetstream.JetStreamConsumer(
return err s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
nats.DeliverAll(), nats.ManualAck(),
)
} }
// onMessage is called when the sync server receives a new event from the client API server output log. // onMessage is called when the sync server receives a new event from the client API server output log.
// It is not safe for this function to be called from multiple goroutines, or else the // It is not safe for this function to be called from multiple goroutines, or else the
// sync stream position may race and be incorrectly calculated. // sync stream position may race and be incorrectly calculated.
func (s *OutputClientDataConsumer) onMessage(msg *nats.Msg) { func (s *OutputClientDataConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
jetstream.WithJetStreamMessage(msg, func(msg *nats.Msg) bool { // Parse out the event JSON
// Parse out the event JSON userID := msg.Header.Get(jetstream.UserID)
userID := msg.Header.Get(jetstream.UserID) var output eventutil.AccountData
var output eventutil.AccountData if err := json.Unmarshal(msg.Data, &output); err != nil {
if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream
// If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("client API server output log: message parse failure")
log.WithError(err).Errorf("client API server output log: message parse failure") sentry.CaptureException(err)
sentry.CaptureException(err)
return true
}
log.WithFields(log.Fields{
"type": output.Type,
"room_id": output.RoomID,
}).Debug("Received data from client API server")
streamPos, err := s.db.UpsertAccountData(
s.ctx, userID, output.RoomID, output.Type,
)
if err != nil {
sentry.CaptureException(err)
log.WithFields(log.Fields{
"type": output.Type,
"room_id": output.RoomID,
log.ErrorKey: err,
}).Panicf("could not save account data")
}
s.stream.Advance(streamPos)
s.notifier.OnNewAccountData(userID, types.StreamingToken{AccountDataPosition: streamPos})
return true return true
}) }
log.WithFields(log.Fields{
"type": output.Type,
"room_id": output.RoomID,
}).Debug("Received data from client API server")
streamPos, err := s.db.UpsertAccountData(
s.ctx, userID, output.RoomID, output.Type,
)
if err != nil {
sentry.CaptureException(err)
log.WithFields(log.Fields{
"type": output.Type,
"room_id": output.RoomID,
log.ErrorKey: err,
}).Panicf("could not save account data")
}
s.stream.Advance(streamPos)
s.notifier.OnNewAccountData(userID, types.StreamingToken{AccountDataPosition: streamPos})
return true
} }

60
syncapi/consumers/eduserver_receipts.go
View File

@ -34,7 +34,7 @@ import (
type OutputReceiptEventConsumer struct { type OutputReceiptEventConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
db storage.Database db storage.Database
stream types.StreamProvider stream types.StreamProvider
@ -64,36 +64,36 @@ func NewOutputReceiptEventConsumer(
// Start consuming from EDU api // Start consuming from EDU api
func (s *OutputReceiptEventConsumer) Start() error { func (s *OutputReceiptEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return jetstream.JetStreamConsumer(
return err s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
nats.DeliverAll(), nats.ManualAck(),
)
} }
func (s *OutputReceiptEventConsumer) onMessage(msg *nats.Msg) { func (s *OutputReceiptEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
jetstream.WithJetStreamMessage(msg, func(msg *nats.Msg) bool { var output api.OutputReceiptEvent
var output api.OutputReceiptEvent if err := json.Unmarshal(msg.Data, &output); err != nil {
if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream
// If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("EDU server output log: message parse failure")
log.WithError(err).Errorf("EDU server output log: message parse failure") sentry.CaptureException(err)
sentry.CaptureException(err)
return true
}
streamPos, err := s.db.StoreReceipt(
s.ctx,
output.RoomID,
output.Type,
output.UserID,
output.EventID,
output.Timestamp,
)
if err != nil {
sentry.CaptureException(err)
return true
}
s.stream.Advance(streamPos)
s.notifier.OnNewReceipt(output.RoomID, types.StreamingToken{ReceiptPosition: streamPos})
return true return true
}) }
streamPos, err := s.db.StoreReceipt(
s.ctx,
output.RoomID,
output.Type,
output.UserID,
output.EventID,
output.Timestamp,
)
if err != nil {
sentry.CaptureException(err)
return true
}
s.stream.Advance(streamPos)
s.notifier.OnNewReceipt(output.RoomID, types.StreamingToken{ReceiptPosition: streamPos})
return true
} }

92
syncapi/consumers/eduserver_sendtodevice.go
View File

@ -36,7 +36,7 @@ import (
type OutputSendToDeviceEventConsumer struct { type OutputSendToDeviceEventConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
db storage.Database db storage.Database
serverName gomatrixserverlib.ServerName // our server name serverName gomatrixserverlib.ServerName // our server name
@ -68,52 +68,52 @@ func NewOutputSendToDeviceEventConsumer(
// Start consuming from EDU api // Start consuming from EDU api
func (s *OutputSendToDeviceEventConsumer) Start() error { func (s *OutputSendToDeviceEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return jetstream.JetStreamConsumer(
return err s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
nats.DeliverAll(), nats.ManualAck(),
)
} }
func (s *OutputSendToDeviceEventConsumer) onMessage(msg *nats.Msg) { func (s *OutputSendToDeviceEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
jetstream.WithJetStreamMessage(msg, func(msg *nats.Msg) bool { var output api.OutputSendToDeviceEvent
var output api.OutputSendToDeviceEvent if err := json.Unmarshal(msg.Data, &output); err != nil {
if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream
// If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("EDU server output log: message parse failure")
log.WithError(err).Errorf("EDU server output log: message parse failure") sentry.CaptureException(err)
sentry.CaptureException(err)
return true
}
_, domain, err := gomatrixserverlib.SplitID('@', output.UserID)
if err != nil {
sentry.CaptureException(err)
return true
}
if domain != s.serverName {
return true
}
util.GetLogger(context.TODO()).WithFields(log.Fields{
"sender": output.Sender,
"user_id": output.UserID,
"device_id": output.DeviceID,
"event_type": output.Type,
}).Info("sync API received send-to-device event from EDU server")
streamPos, err := s.db.StoreNewSendForDeviceMessage(
s.ctx, output.UserID, output.DeviceID, output.SendToDeviceEvent,
)
if err != nil {
sentry.CaptureException(err)
log.WithError(err).Errorf("failed to store send-to-device message")
return false
}
s.stream.Advance(streamPos)
s.notifier.OnNewSendToDevice(
output.UserID,
[]string{output.DeviceID},
types.StreamingToken{SendToDevicePosition: streamPos},
)
return true return true
}) }
_, domain, err := gomatrixserverlib.SplitID('@', output.UserID)
if err != nil {
sentry.CaptureException(err)
return true
}
if domain != s.serverName {
return true
}
util.GetLogger(context.TODO()).WithFields(log.Fields{
"sender": output.Sender,
"user_id": output.UserID,
"device_id": output.DeviceID,
"event_type": output.Type,
}).Info("sync API received send-to-device event from EDU server")
streamPos, err := s.db.StoreNewSendForDeviceMessage(
s.ctx, output.UserID, output.DeviceID, output.SendToDeviceEvent,
)
if err != nil {
sentry.CaptureException(err)
log.WithError(err).Errorf("failed to store send-to-device message")
return false
}
s.stream.Advance(streamPos)
s.notifier.OnNewSendToDevice(
output.UserID,
[]string{output.DeviceID},
types.StreamingToken{SendToDevicePosition: streamPos},
)
return true
} }

70
syncapi/consumers/eduserver_typing.go
View File

@ -35,7 +35,7 @@ import (
type OutputTypingEventConsumer struct { type OutputTypingEventConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
eduCache *cache.EDUCache eduCache *cache.EDUCache
stream types.StreamProvider stream types.StreamProvider
@ -66,41 +66,41 @@ func NewOutputTypingEventConsumer(
// Start consuming from EDU api // Start consuming from EDU api
func (s *OutputTypingEventConsumer) Start() error { func (s *OutputTypingEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return jetstream.JetStreamConsumer(
return err s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
nats.DeliverAll(), nats.ManualAck(),
)
} }
func (s *OutputTypingEventConsumer) onMessage(msg *nats.Msg) { func (s *OutputTypingEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
jetstream.WithJetStreamMessage(msg, func(msg *nats.Msg) bool { var output api.OutputTypingEvent
var output api.OutputTypingEvent if err := json.Unmarshal(msg.Data, &output); err != nil {
if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream
// If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("EDU server output log: message parse failure")
log.WithError(err).Errorf("EDU server output log: message parse failure") sentry.CaptureException(err)
sentry.CaptureException(err)
return true
}
log.WithFields(log.Fields{
"room_id": output.Event.RoomID,
"user_id": output.Event.UserID,
"typing": output.Event.Typing,
}).Debug("received data from EDU server")
var typingPos types.StreamPosition
typingEvent := output.Event
if typingEvent.Typing {
typingPos = types.StreamPosition(
s.eduCache.AddTypingUser(typingEvent.UserID, typingEvent.RoomID, output.ExpireTime),
)
} else {
typingPos = types.StreamPosition(
s.eduCache.RemoveUser(typingEvent.UserID, typingEvent.RoomID),
)
}
s.stream.Advance(typingPos)
s.notifier.OnNewTyping(output.Event.RoomID, types.StreamingToken{TypingPosition: typingPos})
return true return true
}) }
log.WithFields(log.Fields{
"room_id": output.Event.RoomID,
"user_id": output.Event.UserID,
"typing": output.Event.Typing,
}).Debug("received data from EDU server")
var typingPos types.StreamPosition
typingEvent := output.Event
if typingEvent.Typing {
typingPos = types.StreamPosition(
s.eduCache.AddTypingUser(typingEvent.UserID, typingEvent.RoomID, output.ExpireTime),
)
} else {
typingPos = types.StreamPosition(
s.eduCache.RemoveUser(typingEvent.UserID, typingEvent.RoomID),
)
}
s.stream.Advance(typingPos)
s.notifier.OnNewTyping(output.Event.RoomID, types.StreamingToken{TypingPosition: typingPos})
return true
} }

102
syncapi/consumers/roomserver.go
View File

@ -38,7 +38,7 @@ type OutputRoomEventConsumer struct {
cfg *config.SyncAPI cfg *config.SyncAPI
rsAPI api.RoomserverInternalAPI rsAPI api.RoomserverInternalAPI
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
db storage.Database db storage.Database
pduStream types.StreamProvider pduStream types.StreamProvider
@ -73,65 +73,61 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers // Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error { func (s *OutputRoomEventConsumer) Start() error {
_, err := s.jetstream.Subscribe( return jetstream.JetStreamConsumer(
s.topic, s.onMessage, s.durable, s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
nats.DeliverAll(), nats.DeliverAll(), nats.ManualAck(),
nats.ManualAck(),
) )
return err
} }
// onMessage is called when the sync server receives a new event from the room server output log. // onMessage is called when the sync server receives a new event from the room server output log.
// It is not safe for this function to be called from multiple goroutines, or else the // It is not safe for this function to be called from multiple goroutines, or else the
// sync stream position may race and be incorrectly calculated. // sync stream position may race and be incorrectly calculated.
func (s *OutputRoomEventConsumer) onMessage(msg *nats.Msg) { func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
jetstream.WithJetStreamMessage(msg, func(msg *nats.Msg) bool { // Parse out the event JSON
// Parse out the event JSON var err error
var err error var output api.OutputEvent
var output api.OutputEvent if err = json.Unmarshal(msg.Data, &output); err != nil {
if err = json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream
// If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("roomserver output log: message parse failure")
log.WithError(err).Errorf("roomserver output log: message parse failure")
return true
}
switch output.Type {
case api.OutputTypeNewRoomEvent:
// Ignore redaction events. We will add them to the database when they are
// validated (when we receive OutputTypeRedactedEvent)
event := output.NewRoomEvent.Event
if event.Type() == gomatrixserverlib.MRoomRedaction && event.StateKey() == nil {
// in the special case where the event redacts itself, just pass the message through because
// we will never see the other part of the pair
if event.Redacts() != event.EventID() {
return true
}
}
err = s.onNewRoomEvent(s.ctx, *output.NewRoomEvent)
case api.OutputTypeOldRoomEvent:
err = s.onOldRoomEvent(s.ctx, *output.OldRoomEvent)
case api.OutputTypeNewInviteEvent:
s.onNewInviteEvent(s.ctx, *output.NewInviteEvent)
case api.OutputTypeRetireInviteEvent:
s.onRetireInviteEvent(s.ctx, *output.RetireInviteEvent)
case api.OutputTypeNewPeek:
s.onNewPeek(s.ctx, *output.NewPeek)
case api.OutputTypeRetirePeek:
s.onRetirePeek(s.ctx, *output.RetirePeek)
case api.OutputTypeRedactedEvent:
err = s.onRedactEvent(s.ctx, *output.RedactedEvent)
default:
log.WithField("type", output.Type).Debug(
"roomserver output log: ignoring unknown output type",
)
}
if err != nil {
log.WithError(err).Error("roomserver output log: failed to process event")
return false
}
return true return true
}) }
switch output.Type {
case api.OutputTypeNewRoomEvent:
// Ignore redaction events. We will add them to the database when they are
// validated (when we receive OutputTypeRedactedEvent)
event := output.NewRoomEvent.Event
if event.Type() == gomatrixserverlib.MRoomRedaction && event.StateKey() == nil {
// in the special case where the event redacts itself, just pass the message through because
// we will never see the other part of the pair
if event.Redacts() != event.EventID() {
return true
}
}
err = s.onNewRoomEvent(s.ctx, *output.NewRoomEvent)
case api.OutputTypeOldRoomEvent:
err = s.onOldRoomEvent(s.ctx, *output.OldRoomEvent)
case api.OutputTypeNewInviteEvent:
s.onNewInviteEvent(s.ctx, *output.NewInviteEvent)
case api.OutputTypeRetireInviteEvent:
s.onRetireInviteEvent(s.ctx, *output.RetireInviteEvent)
case api.OutputTypeNewPeek:
s.onNewPeek(s.ctx, *output.NewPeek)
case api.OutputTypeRetirePeek:
s.onRetirePeek(s.ctx, *output.RetirePeek)
case api.OutputTypeRedactedEvent:
err = s.onRedactEvent(s.ctx, *output.RedactedEvent)
default:
log.WithField("type", output.Type).Debug(
"roomserver output log: ignoring unknown output type",
)
}
if err != nil {
log.WithError(err).Error("roomserver output log: failed to process event")
return false
}
return true
} }
func (s *OutputRoomEventConsumer) onRedactEvent( func (s *OutputRoomEventConsumer) onRedactEvent(