bugfix: fix race condition when updating presence via /sync (#2470)

* bugfix: fix race condition when updating presence via /sync

Previously when presence is updated via /sync, we would send the presence update
asyncly via NATS. This created a race condition:
 - If the presence update is processed quickly, the /sync which triggered the presence
   update would see an online presence.
 - If the presence update was processed slowly, the /sync which triggered the presence
   update would see an offline presence.

This is the root cause behind the flakey sytest: 'User sees their own presence in a sync'.

The fix is to ensure we update the database/advance the stream position synchronously
for local users.

* Bugfix for test

1 files changed, 55 insertions, 0 deletions

diff --git a/syncapi/syncapi_test.go b/syncapi/syncapi_test.go
index 5ecfd877..3ce7c64b 100644
--- a/syncapi/syncapi_test.go
+++ b/syncapi/syncapi_test.go
@@ -19,6 +19,7 @@ import (
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/nats-io/nats.go"
+	"github.com/tidwall/gjson"
 )
 
 type syncRoomserverAPI struct {
@@ -256,6 +257,60 @@ func testSyncAPICreateRoomSyncEarly(t *testing.T, dbType test.DBType) {
 	}
 }
 
+// Test that if we hit /sync we get back presence: online, regardless of whether messages get delivered
+// via NATS. Regression test for a flakey test "User sees their own presence in a sync"
+func TestSyncAPIUpdatePresenceImmediately(t *testing.T) {
+	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+		testSyncAPIUpdatePresenceImmediately(t, dbType)
+	})
+}
+
+func testSyncAPIUpdatePresenceImmediately(t *testing.T, dbType test.DBType) {
+	user := test.NewUser(t)
+	alice := userapi.Device{
+		ID:          "ALICEID",
+		UserID:      user.ID,
+		AccessToken: "ALICE_BEARER_TOKEN",
+		DisplayName: "Alice",
+		AccountType: userapi.AccountTypeUser,
+	}
+
+	base, close := testrig.CreateBaseDendrite(t, dbType)
+	base.Cfg.Global.Presence.EnableOutbound = true
+	base.Cfg.Global.Presence.EnableInbound = true
+	defer close()
+
+	jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
+	defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
+	AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{}, &syncKeyAPI{})
+	w := httptest.NewRecorder()
+	base.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
+		"access_token": alice.AccessToken,
+		"timeout":      "0",
+		"set_presence": "online",
+	})))
+	if w.Code != 200 {
+		t.Fatalf("got HTTP %d want %d", w.Code, 200)
+	}
+	var res types.Response
+	if err := json.NewDecoder(w.Body).Decode(&res); err != nil {
+		t.Errorf("failed to decode response body: %s", err)
+	}
+	if len(res.Presence.Events) != 1 {
+		t.Fatalf("expected 1 presence events, got: %+v", res.Presence.Events)
+	}
+	if res.Presence.Events[0].Sender != alice.UserID {
+		t.Errorf("sender: got %v want %v", res.Presence.Events[0].Sender, alice.UserID)
+	}
+	if res.Presence.Events[0].Type != "m.presence" {
+		t.Errorf("type: got %v want %v", res.Presence.Events[0].Type, "m.presence")
+	}
+	if gjson.ParseBytes(res.Presence.Events[0].Content).Get("presence").Str != "online" {
+		t.Errorf("content: not online,  got %v", res.Presence.Events[0].Content)
+	}
+
+}
+
 func toNATSMsgs(t *testing.T, base *base.BaseDendrite, input []*gomatrixserverlib.HeaderedEvent) []*nats.Msg {
 	result := make([]*nats.Msg, len(input))
 	for i, ev := range input {