Clearer contexts, don't use background in case there's something happening there

This commit is contained in:
Neil Alexander 2022-01-26 16:30:43 +00:00
parent 8813ec6066
commit 3150f6e64e
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
3 changed files with 15 additions and 16 deletions

View file

@ -164,7 +164,7 @@ func Send(
util.GetLogger(httpReq.Context()).Infof("Received transaction %q from %q containing %d PDUs, %d EDUs", txnID, request.Origin(), len(t.PDUs), len(t.EDUs)) util.GetLogger(httpReq.Context()).Infof("Received transaction %q from %q containing %d PDUs, %d EDUs", txnID, request.Origin(), len(t.PDUs), len(t.EDUs))
resp, jsonErr := t.processTransaction(context.Background()) resp, jsonErr := t.processTransaction(httpReq.Context())
if jsonErr != nil { if jsonErr != nil {
util.GetLogger(httpReq.Context()).WithField("jsonErr", jsonErr).Error("t.processTransaction failed") util.GetLogger(httpReq.Context()).WithField("jsonErr", jsonErr).Error("t.processTransaction failed")
return *jsonErr return *jsonErr
@ -276,7 +276,7 @@ func (t *txnReq) processTransaction(ctx context.Context) (*gomatrixserverlib.Res
// If the event fail auth checks, gmsl.NotAllowed error will be returned which we be silently // If the event fail auth checks, gmsl.NotAllowed error will be returned which we be silently
// discarded by the caller of this function // discarded by the caller of this function
if err = api.SendEvents( if err = api.SendEvents(
context.Background(), ctx,
t.rsAPI, t.rsAPI,
api.KindNew, api.KindNew,
[]*gomatrixserverlib.HeaderedEvent{ []*gomatrixserverlib.HeaderedEvent{

View file

@ -65,24 +65,25 @@ var processRoomEventDuration = prometheus.NewHistogramVec(
// TODO: Break up function - we should probably do transaction ID checks before calling this. // TODO: Break up function - we should probably do transaction ID checks before calling this.
// nolint:gocyclo // nolint:gocyclo
func (r *Inputer) processRoomEvent( func (r *Inputer) processRoomEvent(
ctx context.Context, inctx context.Context,
input *api.InputRoomEvent, input *api.InputRoomEvent,
) (err error) { ) (err error) {
select { select {
case <-ctx.Done(): case <-inctx.Done():
// Before we do anything, make sure the context hasn't expired for this pending task. // Before we do anything, make sure the context hasn't expired for this pending task.
// If it has then we'll give up straight away — it's probably a synchronous input // If it has then we'll give up straight away — it's probably a synchronous input
// request and the caller has already given up, but the inbox task was still queued. // request and the caller has already given up, but the inbox task was still queued.
return context.DeadlineExceeded return context.DeadlineExceeded
default: default:
// Otherwise we're going to wrap the context with a time limit. We'll allow no more
// than MaximumProcessingTime for everything that we need to do for this event, or
// it's possible that we could end up wedging the roomserver for a very long time.
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, MaximumProcessingTime)
defer cancel()
} }
// Wrap the context with a time limit. We'll allow no more than MaximumProcessingTime for
// everything that we need to do for this event, or it's possible that we could end up wedging
// the roomserver for a very long time.
var cancel context.CancelFunc
ctx, cancel := context.WithTimeout(inctx, MaximumProcessingTime)
defer cancel()
// Measure how long it takes to process this event. // Measure how long it takes to process this event.
started := time.Now() started := time.Now()
defer func() { defer func() {

View file

@ -394,9 +394,7 @@ func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserve
var missingResp *gomatrixserverlib.RespMissingEvents var missingResp *gomatrixserverlib.RespMissingEvents
for server := range t.servers { for server := range t.servers {
var m gomatrixserverlib.RespMissingEvents var m gomatrixserverlib.RespMissingEvents
reqctx, cancel := context.WithTimeout(ctx, time.Second*30) if m, err = t.federation.LookupMissingEvents(ctx, server, e.RoomID(), gomatrixserverlib.MissingEvents{
defer cancel()
if m, err = t.federation.LookupMissingEvents(reqctx, server, e.RoomID(), gomatrixserverlib.MissingEvents{
Limit: 20, Limit: 20,
// The latest event IDs that the sender already has. These are skipped when retrieving the previous events of latest_events. // The latest event IDs that the sender already has. These are skipped when retrieving the previous events of latest_events.
EarliestEvents: latestEvents, EarliestEvents: latestEvents,
@ -409,10 +407,10 @@ func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserve
logger.WithError(err).Errorf("%s pushed us an event but %q did not respond to /get_missing_events", t.origin, server) logger.WithError(err).Errorf("%s pushed us an event but %q did not respond to /get_missing_events", t.origin, server)
if errors.Is(err, context.DeadlineExceeded) { if errors.Is(err, context.DeadlineExceeded) {
select { select {
case <-reqctx.Done(): // this server took too long case <-ctx.Done(): // the parent request context timed out
continue
case <-ctx.Done(): // the input request timed out
return nil, context.DeadlineExceeded return nil, context.DeadlineExceeded
default: // this request exceed its own timeout
continue
} }
} }
} }