processRoomEvent can only be called once at a time TODO(#375): This should be rewritten to allow concurrent calls. The difficulty is in ensuring that we correctly annotate events with the correct state deltas when sending to kafka streams TODO: Break up function - we should probably do transaction I
( // nats-server ctx context.Context, input *api.InputRoomEvent, )
| 64 | // TODO: Break up function - we should probably do transaction ID checks before calling this. |
| 65 | // nolint:gocyclo |
| 66 | func (r *Inputer) processRoomEvent( // nats-server |
| 67 | ctx context.Context, |
| 68 | input *api.InputRoomEvent, |
| 69 | ) error { |
| 70 | select { |
| 71 | case <-ctx.Done(): |
| 72 | // Before we do anything, make sure the context hasn't expired for this pending task. |
| 73 | // If it has then we'll give up straight away — it's probably a synchronous input |
| 74 | // request and the caller has already given up, but the inbox task was still queued. |
| 75 | return context.DeadlineExceeded |
| 76 | default: |
| 77 | } |
| 78 | |
| 79 | span, ctx := opentracing.StartSpanFromContext(ctx, "processRoomEvent") |
| 80 | span.SetTag("room_id", input.Event.RoomID()) |
| 81 | span.SetTag("event_id", input.Event.EventID()) |
| 82 | defer span.Finish() |
| 83 | |
| 84 | // Measure how long it takes to process this event. |
| 85 | started := time.Now() |
| 86 | defer func() { |
| 87 | timetaken := time.Since(started) |
| 88 | processRoomEventDuration.With(prometheus.Labels{ |
| 89 | "room_id": input.Event.RoomID(), |
| 90 | }).Observe(float64(timetaken.Milliseconds())) |
| 91 | }() |
| 92 | |
| 93 | // Parse and validate the event JSON |
| 94 | headered := input.Event |
| 95 | event := headered.Unwrap() |
| 96 | logger := util.GetLogger(ctx).WithFields(logrus.Fields{ |
| 97 | "event_id": event.EventID(), |
| 98 | "room_id": event.RoomID(), |
| 99 | "kind": input.Kind, |
| 100 | "origin": input.Origin, |
| 101 | "type": event.Type(), |
| 102 | }) |
| 103 | if input.HasState { |
| 104 | logger = logger.WithFields(logrus.Fields{ |
| 105 | "has_state": input.HasState, |
| 106 | "state_ids": len(input.StateEventIDs), |
| 107 | }) |
| 108 | } |
| 109 | |
| 110 | // if we have already got this event then do not process it again, if the input kind is an outlier. |
| 111 | // Outliers contain no extra information which may warrant a re-processing. |
| 112 | if input.Kind == api.KindOutlier { |
| 113 | evs, err2 := r.DB.EventsFromIDs(ctx, []string{event.EventID()}) |
| 114 | if err2 == nil && len(evs) == 1 { |
| 115 | // check hash matches if we're on early room versions where the event ID was a random string |
| 116 | idFormat, err2 := headered.RoomVersion.EventIDFormat() |
| 117 | if err2 == nil { |
| 118 | switch idFormat { |
| 119 | case gomatrixserverlib.EventIDFormatV1: |
| 120 | if bytes.Equal(event.EventReference().EventSHA256, evs[0].EventReference().EventSHA256) { |
| 121 | logger.Debugf("Already processed event; ignoring") |
| 122 | return nil |
| 123 | } |
no test coverage detected