Additional tests; linting

internal/errors.go

@@ -128,6 +128,6 @@ func (e *DataError) Error() string {
 
 func NewDataError(msg string, args ...interface{}) error {
 	return &DataError{
-		msg: fmt.Sprintf(msg, args...),
+		msg: fmt.Sprintf("DataError: "+msg, args...),
 	}
 }

sync2/poller.go

@@ -680,10 +680,7 @@ func shouldRetry(retryErr error) bool {
 	}
 	// we retry on all errors EXCEPT DataError as this indicates that retrying won't help
 	var de *internal.DataError
-	if errors.As(retryErr, &de) {
-		return false
-	}
-	return true
+	return !errors.As(retryErr, &de)
 }
 
 func (p *poller) parseToDeviceMessages(ctx context.Context, res *SyncResponse) error {

tests-integration/poller_test.go

@@ -4,13 +4,14 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/jmoiron/sqlx"
-	"github.com/matrix-org/sliding-sync/sqlutil"
 	"net/http"
 	"os"
 	"testing"
 	"time"
 
+	"github.com/jmoiron/sqlx"
+	"github.com/matrix-org/sliding-sync/sqlutil"
+
 	"github.com/matrix-org/sliding-sync/sync2"
 	"github.com/matrix-org/sliding-sync/sync3"
 	"github.com/matrix-org/sliding-sync/sync3/extensions"
@@ -415,7 +416,7 @@ func TestPollersCanBeResumedAfterExpiry(t *testing.T) {
 	res := v3.mustDoV3Request(t, aliceToken, sync3.Request{
 		Extensions: extensions.Request{
 			AccountData: &extensions.AccountDataRequest{
-				extensions.Core{
+				Core: extensions.Core{
 					Enabled: &boolTrue,
 				},
 			},

tests-integration/regressions_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/matrix-org/sliding-sync/sync2"
 	"github.com/matrix-org/sliding-sync/sync3"
+	"github.com/matrix-org/sliding-sync/sync3/extensions"
 	"github.com/matrix-org/sliding-sync/testutils"
 	"github.com/matrix-org/sliding-sync/testutils/m"
 )
@@ -343,5 +344,122 @@ func TestBadCreateInitialiseDoesntWedgePolling(t *testing.T) {
 	case <-time.After(time.Second):
 		t.Fatalf("timed out waiting for all v2 requests")
 	}
-
+}
+
+// Regression test for https://github.com/matrix-org/sliding-sync/issues/295
+// This test:
+// - injects to-device msgs and a bad room in the v2 response
+// - checks we see the to-device msgs
+// It's possible for the poller to bail early and skip over processing the remaining response
+// which this test is trying to safeguard against.
+func TestBadPollDataDoesntDropToDeviceMsgs(t *testing.T) {
+	pqString := testutils.PrepareDBConnectionString()
+	// setup code
+	v2 := runTestV2Server(t)
+	v3 := runTestServer(t, v2, pqString)
+	defer v2.close()
+	defer v3.close()
+
+	goodRoom := "!good:localhost"
+	badRoom := "!bad:localhost"
+
+	v2.addAccount(t, alice, aliceToken)
+	// we should see the since token increment, if we see repeats it means
+	// we aren't returning DataErrors when we should be.
+	wantSinces := []string{"", "1", "2"}
+	ch := make(chan bool)
+	v2.checkRequest = func(token string, req *http.Request) {
+		if len(wantSinces) == 0 {
+			return
+		}
+		gotSince := req.URL.Query().Get("since")
+		t.Logf("checkRequest got since=%v", gotSince)
+		want := wantSinces[0]
+		wantSinces = wantSinces[1:]
+		if gotSince != want {
+			t.Errorf("v2.checkRequest since got '%v' want '%v'", gotSince, want)
+		}
+		if len(wantSinces) == 0 {
+			close(ch)
+		}
+	}
+
+	// initial sync, everything fine
+	v2.queueResponse(alice, sync2.SyncResponse{
+		NextBatch: "1",
+		Rooms: sync2.SyncRoomsResponse{
+			Join: map[string]sync2.SyncV2JoinResponse{
+				goodRoom: {
+					Timeline: sync2.TimelineResponse{
+						Events: createRoomState(t, alice, time.Now()),
+					},
+				},
+			},
+		},
+	})
+	aliceRes := v3.mustDoV3Request(t, aliceToken, sync3.Request{
+		Extensions: extensions.Request{
+			ToDevice: &extensions.ToDeviceRequest{
+				Core: extensions.Core{
+					Enabled: &boolTrue,
+				},
+			},
+		},
+		Lists: map[string]sync3.RequestList{
+			"a": {
+				Ranges: sync3.SliceRanges{{0, 20}},
+				RoomSubscription: sync3.RoomSubscription{
+					TimelineLimit: 1,
+				},
+			},
+		},
+	})
+	// we should only see 1 room
+	m.MatchResponse(t, aliceRes, m.MatchRoomSubscriptionsStrict(map[string][]m.RoomMatcher{
+		goodRoom: {
+			m.MatchJoinCount(1),
+		},
+	},
+	))
+
+	// now inject a bad room and some to-device events
+	toDeviceEvent := testutils.NewEvent(t, "m.todevice", alice, map[string]interface{}{"body": "testio"})
+	v2.queueResponse(alice, sync2.SyncResponse{
+		NextBatch: "2",
+		ToDevice: sync2.EventsResponse{
+			Events: []json.RawMessage{toDeviceEvent},
+		},
+		Rooms: sync2.SyncRoomsResponse{
+			Join: map[string]sync2.SyncV2JoinResponse{
+				badRoom: {
+					State: sync2.EventsResponse{
+						// BAD: missing create event
+						Events: createRoomState(t, alice, time.Now())[1:],
+					},
+					Timeline: sync2.TimelineResponse{
+						Events: []json.RawMessage{
+							testutils.NewMessageEvent(t, alice, "Hello World"),
+						},
+					},
+				},
+			},
+		},
+	})
+	v2.waitUntilEmpty(t, alice)
+
+	// we should see the to-device event and not the bad room
+	aliceRes = v3.mustDoV3RequestWithPos(t, aliceToken, aliceRes.Pos, sync3.Request{})
+	// we should only see the to-device event
+	m.MatchResponse(t, aliceRes,
+		m.LogResponse(t),
+		m.MatchRoomSubscriptionsStrict(map[string][]m.RoomMatcher{}),
+		m.MatchToDeviceMessages([]json.RawMessage{toDeviceEvent}),
+	)
+
+	// make sure we've seen all the v2 requests
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		t.Fatalf("timed out waiting for all v2 requests")
+	}
 }