get missing events for `e`. If `isGapFilled`=true then `newEvents` contains all the events to inject, without `e`. If `isGapFilled=false` then `newEvents` contains the response to /get_missing_events
(ctx context.Context, e *gomatrixserverlib.Event, roomVersion gomatrixserverlib.RoomVersion)
| 501 | // get missing events for `e`. If `isGapFilled`=true then `newEvents` contains all the events to inject, |
| 502 | // without `e`. If `isGapFilled=false` then `newEvents` contains the response to /get_missing_events |
| 503 | func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserverlib.Event, roomVersion gomatrixserverlib.RoomVersion) (newEvents []*gomatrixserverlib.Event, isGapFilled, prevStateKnown bool, err error) { |
| 504 | span, ctx := opentracing.StartSpanFromContext(ctx, "getMissingEvents") |
| 505 | defer span.Finish() |
| 506 | |
| 507 | logger := util.GetLogger(ctx).WithField("event_id", e.EventID()).WithField("room_id", e.RoomID()) |
| 508 | latest, _, _, err := t.db.LatestEventIDs(ctx, t.roomInfo.RoomNID) |
| 509 | if err != nil { |
| 510 | return nil, false, false, fmt.Errorf("t.DB.LatestEventIDs: %w", err) |
| 511 | } |
| 512 | latestEvents := make([]string, len(latest)) |
| 513 | for i, ev := range latest { |
| 514 | latestEvents[i] = ev.EventID |
| 515 | t.hadEvent(ev.EventID) |
| 516 | } |
| 517 | |
| 518 | var missingResp *gomatrixserverlib.RespMissingEvents |
| 519 | for _, server := range t.servers { |
| 520 | var m gomatrixserverlib.RespMissingEvents |
| 521 | if m, err = t.federation.LookupMissingEvents(ctx, server, e.RoomID(), gomatrixserverlib.MissingEvents{ |
| 522 | Limit: 20, |
| 523 | // The latest event IDs that the sender already has. These are skipped when retrieving the previous events of latest_events. |
| 524 | EarliestEvents: latestEvents, |
| 525 | // The event IDs to retrieve the previous events for. |
| 526 | LatestEvents: []string{e.EventID()}, |
| 527 | }, roomVersion); err == nil { |
| 528 | missingResp = &m |
| 529 | break |
| 530 | } else { |
| 531 | logger.WithError(err).Warnf("%s pushed us an event but %q did not respond to /get_missing_events", t.origin, server) |
| 532 | if errors.Is(err, context.DeadlineExceeded) { |
| 533 | select { |
| 534 | case <-ctx.Done(): // the parent request context timed out |
| 535 | return nil, false, false, context.DeadlineExceeded |
| 536 | default: // this request exceed its own timeout |
| 537 | continue |
| 538 | } |
| 539 | } |
| 540 | } |
| 541 | } |
| 542 | |
| 543 | if missingResp == nil { |
| 544 | logger.WithError(err).Warnf( |
| 545 | "%s pushed us an event but %d server(s) couldn't give us details about prev_events via /get_missing_events - dropping this event until it can", |
| 546 | t.origin, len(t.servers), |
| 547 | ) |
| 548 | return nil, false, false, missingPrevEventsError{ |
| 549 | eventID: e.EventID(), |
| 550 | err: err, |
| 551 | } |
| 552 | } |
| 553 | |
| 554 | // Make sure events from the missingResp are using the cache - missing events |
| 555 | // will be added and duplicates will be removed. |
| 556 | logger.Debugf("get_missing_events returned %d events", len(missingResp.Events)) |
| 557 | missingEvents := make([]*gomatrixserverlib.Event, 0, len(missingResp.Events)) |
| 558 | for _, ev := range missingResp.Events.UntrustedEvents(roomVersion) { |
| 559 | missingEvents = append(missingEvents, t.cacheAndReturn(ev)) |
| 560 | } |