Eliminate more SQL no row errors in sync API (#2363)

* Handle `sql.ErrNoRows` in main `/sync` codepaths

* Catch more

2 files changed, 37 insertions, 4 deletions

diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go
index 91eba44e..2143fd67 100644
--- a/syncapi/storage/shared/syncserver.go
+++ b/syncapi/storage/shared/syncserver.go
@@ -688,6 +688,9 @@ func (d *Database) GetStateDeltas(
 	// user has ever interacted with — joined to, kicked/banned from, left.
 	memberships, err := d.CurrentRoomState.SelectRoomIDsWithAnyMembership(ctx, txn, userID)
 	if err != nil {
+		if err == sql.ErrNoRows {
+			return nil, nil, nil
+		}
 		return nil, nil, err
 	}
 
@@ -705,17 +708,23 @@ func (d *Database) GetStateDeltas(
 	// get all the state events ever (i.e. for all available rooms) between these two positions
 	stateNeeded, eventMap, err := d.OutputEvents.SelectStateInRange(ctx, txn, r, stateFilter, allRoomIDs)
 	if err != nil {
+		if err == sql.ErrNoRows {
+			return nil, nil, nil
+		}
 		return nil, nil, err
 	}
 	state, err := d.fetchStateEvents(ctx, txn, stateNeeded, eventMap)
 	if err != nil {
+		if err == sql.ErrNoRows {
+			return nil, nil, nil
+		}
 		return nil, nil, err
 	}
 
 	// find out which rooms this user is peeking, if any.
 	// We do this before joins so any peeks get overwritten
 	peeks, err := d.Peeks.SelectPeeksInRange(ctx, txn, userID, device.ID, r)
-	if err != nil {
+	if err != nil && err != sql.ErrNoRows {
 		return nil, nil, err
 	}
 
@@ -726,6 +735,9 @@ func (d *Database) GetStateDeltas(
 			var s []types.StreamEvent
 			s, err = d.currentStateStreamEventsForRoom(ctx, txn, peek.RoomID, stateFilter)
 			if err != nil {
+				if err == sql.ErrNoRows {
+					continue
+				}
 				return nil, nil, err
 			}
 			state[peek.RoomID] = s
@@ -753,6 +765,9 @@ func (d *Database) GetStateDeltas(
 					var s []types.StreamEvent
 					s, err = d.currentStateStreamEventsForRoom(ctx, txn, roomID, stateFilter)
 					if err != nil {
+						if err == sql.ErrNoRows {
+							continue
+						}
 						return nil, nil, err
 					}
 					state[roomID] = s
@@ -803,6 +818,9 @@ func (d *Database) GetStateDeltasForFullStateSync(
 	// user has ever interacted with — joined to, kicked/banned from, left.
 	memberships, err := d.CurrentRoomState.SelectRoomIDsWithAnyMembership(ctx, txn, userID)
 	if err != nil {
+		if err == sql.ErrNoRows {
+			return nil, nil, nil
+		}
 		return nil, nil, err
 	}
 
@@ -819,7 +837,7 @@ func (d *Database) GetStateDeltasForFullStateSync(
 	deltas := make(map[string]types.StateDelta)
 
 	peeks, err := d.Peeks.SelectPeeksInRange(ctx, txn, userID, device.ID, r)
-	if err != nil {
+	if err != nil && err != sql.ErrNoRows {
 		return nil, nil, err
 	}
 
@@ -828,6 +846,9 @@ func (d *Database) GetStateDeltasForFullStateSync(
 		if !peek.Deleted {
 			s, stateErr := d.currentStateStreamEventsForRoom(ctx, txn, peek.RoomID, stateFilter)
 			if stateErr != nil {
+				if stateErr == sql.ErrNoRows {
+					continue
+				}
 				return nil, nil, stateErr
 			}
 			deltas[peek.RoomID] = types.StateDelta{
@@ -841,10 +862,16 @@ func (d *Database) GetStateDeltasForFullStateSync(
 	// Get all the state events ever between these two positions
 	stateNeeded, eventMap, err := d.OutputEvents.SelectStateInRange(ctx, txn, r, stateFilter, allRoomIDs)
 	if err != nil {
+		if err == sql.ErrNoRows {
+			return nil, nil, nil
+		}
 		return nil, nil, err
 	}
 	state, err := d.fetchStateEvents(ctx, txn, stateNeeded, eventMap)
 	if err != nil {
+		if err == sql.ErrNoRows {
+			return nil, nil, nil
+		}
 		return nil, nil, err
 	}
 
@@ -869,6 +896,9 @@ func (d *Database) GetStateDeltasForFullStateSync(
 	for _, joinedRoomID := range joinedRoomIDs {
 		s, stateErr := d.currentStateStreamEventsForRoom(ctx, txn, joinedRoomID, stateFilter)
 		if stateErr != nil {
+			if stateErr == sql.ErrNoRows {
+				continue
+			}
 			return nil, nil, stateErr
 		}
 		deltas[joinedRoomID] = types.StateDelta{
diff --git a/syncapi/streams/stream_pdu.go b/syncapi/streams/stream_pdu.go
index 0f11d55f..df5fb8e0 100644
--- a/syncapi/streams/stream_pdu.go
+++ b/syncapi/streams/stream_pdu.go
@@ -341,12 +341,16 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
 	wantFullState bool,
 	device *userapi.Device,
 ) (jr *types.JoinResponse, err error) {
+	jr = types.NewJoinResponse()
 	// TODO: When filters are added, we may need to call this multiple times to get enough events.
 	//       See: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L316
 	recentStreamEvents, limited, err := p.DB.RecentEvents(
 		ctx, roomID, r, eventFilter, true, true,
 	)
 	if err != nil {
+		if err == sql.ErrNoRows {
+			return jr, nil
+		}
 		return
 	}
 
@@ -430,12 +434,11 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
 			false, limited, stateFilter.IncludeRedundantMembers,
 			device, recentEvents, stateEvents,
 		)
-		if err != nil {
+		if err != nil && err != sql.ErrNoRows {
 			return nil, err
 		}
 	}
 
-	jr = types.NewJoinResponse()
 	jr.Summary.JoinedMemberCount = &joinedCount
 	jr.Summary.InvitedMemberCount = &invitedCount
 	jr.Timeline.PrevBatch = prevBatch