diff --git a/internal/client/client.go b/internal/client/client.go index 83d7772d..9a52510d 100644 --- a/internal/client/client.go +++ b/internal/client/client.go @@ -15,6 +15,7 @@ import ( "net/url" "strconv" "strings" + "sync/atomic" "testing" "time" @@ -93,7 +94,7 @@ type CSAPI struct { // True to enable verbose logging Debug bool - txnID int + txnID int64 } // UploadContent uploads the provided content with an optional file name. Fails the test on error. Returns the MXC URI. @@ -264,8 +265,8 @@ func (c *CSAPI) SetPushRule(t *testing.T, scope string, kind string, ruleID stri // Returns the event ID of the sent event. func (c *CSAPI) SendEventUnsynced(t *testing.T, roomID string, e b.Event) string { t.Helper() - c.txnID++ - paths := []string{"_matrix", "client", "v3", "rooms", roomID, "send", e.Type, strconv.Itoa(c.txnID)} + txnID := int(atomic.AddInt64(&c.txnID, 1)) + paths := []string{"_matrix", "client", "v3", "rooms", roomID, "send", e.Type, strconv.Itoa(txnID)} if e.StateKey != nil { paths = []string{"_matrix", "client", "v3", "rooms", roomID, "state", e.Type, *e.StateKey} } @@ -290,8 +291,8 @@ func (c *CSAPI) SendEventSynced(t *testing.T, roomID string, e b.Event) string { // SendRedaction sends a redaction request. Will fail if the returned HTTP request code is not 200 func (c *CSAPI) SendRedaction(t *testing.T, roomID string, e b.Event, eventID string) string { t.Helper() - c.txnID++ - paths := []string{"_matrix", "client", "v3", "rooms", roomID, "redact", eventID, strconv.Itoa(c.txnID)} + txnID := int(atomic.AddInt64(&c.txnID, 1)) + paths := []string{"_matrix", "client", "v3", "rooms", roomID, "redact", eventID, strconv.Itoa(txnID)} res := c.MustDoFunc(t, "PUT", paths, WithJSONBody(t, e.Content)) body := ParseJSON(t, res) return GetJSONFieldStr(t, body, "event_id") @@ -1008,11 +1009,11 @@ func SplitMxc(mxcUri string) (string, string) { // user_id -> device_id -> content (map[string]interface{}) func (c *CSAPI) SendToDeviceMessages(t *testing.T, evType string, messages map[string]map[string]map[string]interface{}) { t.Helper() - c.txnID++ + txnID := int(atomic.AddInt64(&c.txnID, 1)) c.MustDoFunc( t, "PUT", - []string{"_matrix", "client", "v3", "sendToDevice", evType, strconv.Itoa(c.txnID)}, + []string{"_matrix", "client", "v3", "sendToDevice", evType, strconv.Itoa(txnID)}, WithJSONBody( t, map[string]map[string]map[string]map[string]interface{}{ diff --git a/internal/docker/deployment.go b/internal/docker/deployment.go index effcd523..cefb2404 100644 --- a/internal/docker/deployment.go +++ b/internal/docker/deployment.go @@ -1,6 +1,7 @@ package docker import ( + "sync" "testing" "time" @@ -26,9 +27,11 @@ type HomeserverDeployment struct { FedBaseURL string // e.g https://localhost:48373 ContainerID string // e.g 10de45efba AccessTokens map[string]string // e.g { "@alice:hs1": "myAcc3ssT0ken" } + accessTokensMutex sync.RWMutex ApplicationServices map[string]string // e.g { "my-as-id": "id: xxx\nas_token: xxx ..."} } DeviceIDs map[string]string // e.g { "@alice:hs1": "myDeviceID" } CSAPIClients []*client.CSAPI + CSAPIClientsMutex sync.Mutex } // Updates the client and federation base URLs of the homeserver deployment. @@ -58,7 +61,9 @@ func (d *Deployment) Client(t *testing.T, hsName, userID string) *client.CSAPI { t.Fatalf("Deployment.Client - HS name '%s' not found", hsName) return nil } + dep.accessTokensMutex.RLock() token := dep.AccessTokens[userID] + dep.accessTokensMutex.RUnlock() if token == "" && userID != "" { t.Fatalf("Deployment.Client - HS name '%s' - user ID '%s' not found", hsName, userID) return nil @@ -76,7 +81,11 @@ func (d *Deployment) Client(t *testing.T, hsName, userID string) *client.CSAPI { SyncUntilTimeout: 5 * time.Second, Debug: d.Deployer.debugLogging, } + // Appending a slice is not thread-safe. Protect the write with a mutex. + dep.CSAPIClientsMutex.Lock() dep.CSAPIClients = append(dep.CSAPIClients, client) + dep.CSAPIClientsMutex.Unlock() + return client } @@ -101,7 +110,10 @@ func (d *Deployment) RegisterUser(t *testing.T, hsName, localpart, password stri SyncUntilTimeout: 5 * time.Second, Debug: d.Deployer.debugLogging, } + // Appending a slice is not thread-safe. Protect the write with a mutex. + dep.CSAPIClientsMutex.Lock() dep.CSAPIClients = append(dep.CSAPIClients, client) + dep.CSAPIClientsMutex.Unlock() var userID, accessToken, deviceID string if isAdmin { userID, accessToken, deviceID = client.RegisterSharedSecret(t, localpart, password, isAdmin) @@ -110,7 +122,9 @@ func (d *Deployment) RegisterUser(t *testing.T, hsName, localpart, password stri } // remember the token so subsequent calls to deployment.Client return the user + dep.accessTokensMutex.Lock() dep.AccessTokens[userID] = accessToken + dep.accessTokensMutex.Unlock() client.UserID = userID client.AccessToken = accessToken @@ -133,7 +147,10 @@ func (d *Deployment) LoginUser(t *testing.T, hsName, localpart, password string) SyncUntilTimeout: 5 * time.Second, Debug: d.Deployer.debugLogging, } + // Appending a slice is not thread-safe. Protect the write with a mutex. + dep.CSAPIClientsMutex.Lock() dep.CSAPIClients = append(dep.CSAPIClients, client) + dep.CSAPIClientsMutex.Unlock() userID, accessToken, deviceID := client.LoginUser(t, localpart, password) client.UserID = userID diff --git a/internal/federation/handle.go b/internal/federation/handle.go index dd8cc12b..c6374495 100644 --- a/internal/federation/handle.go +++ b/internal/federation/handle.go @@ -163,6 +163,7 @@ func SendJoinRequestsHandler(s *Server, w http.ResponseWriter, req *http.Request // build the state list *before* we insert the new event var stateEvents []*gomatrixserverlib.Event + room.StateMutex.RLock() for _, ev := range room.State { // filter out non-critical memberships if this is a partial-state join if expectPartialState { @@ -172,6 +173,7 @@ func SendJoinRequestsHandler(s *Server, w http.ResponseWriter, req *http.Request } stateEvents = append(stateEvents, ev) } + room.StateMutex.RUnlock() authEvents := room.AuthChainForEvents(stateEvents) diff --git a/internal/federation/server_room.go b/internal/federation/server_room.go index 5c9dad68..c17d94ab 100644 --- a/internal/federation/server_room.go +++ b/internal/federation/server_room.go @@ -3,6 +3,7 @@ package federation import ( "encoding/json" "fmt" + "sync" "testing" "github.com/matrix-org/gomatrixserverlib" @@ -15,7 +16,9 @@ type ServerRoom struct { Version gomatrixserverlib.RoomVersion RoomID string State map[string]*gomatrixserverlib.Event + StateMutex sync.RWMutex Timeline []*gomatrixserverlib.Event + TimelineMutex sync.RWMutex ForwardExtremities []string Depth int64 } @@ -36,7 +39,9 @@ func (r *ServerRoom) AddEvent(ev *gomatrixserverlib.Event) { if ev.StateKey() != nil { r.replaceCurrentState(ev) } + r.TimelineMutex.Lock() r.Timeline = append(r.Timeline, ev) + r.TimelineMutex.Unlock() // update extremities and depth if ev.Depth() > r.Depth { r.Depth = ev.Depth() @@ -75,20 +80,27 @@ func (r *ServerRoom) AuthEvents(sn gomatrixserverlib.StateNeeded) (eventIDs []st // on the (type, state_key) provided. func (r *ServerRoom) replaceCurrentState(ev *gomatrixserverlib.Event) { tuple := fmt.Sprintf("%s\x1f%s", ev.Type(), *ev.StateKey()) + r.StateMutex.Lock() r.State[tuple] = ev + r.StateMutex.Unlock() } // CurrentState returns the state event for the given (type, state_key) or nil. func (r *ServerRoom) CurrentState(evType, stateKey string) *gomatrixserverlib.Event { tuple := fmt.Sprintf("%s\x1f%s", evType, stateKey) - return r.State[tuple] + r.StateMutex.RLock() + state := r.State[tuple] + r.StateMutex.RUnlock() + return state } // AllCurrentState returns all the current state events func (r *ServerRoom) AllCurrentState() (events []*gomatrixserverlib.Event) { + r.StateMutex.RLock() for _, ev := range r.State { events = append(events, ev) } + r.StateMutex.RUnlock() return } @@ -104,12 +116,16 @@ func (r *ServerRoom) AuthChainForEvents(events []*gomatrixserverlib.Event) (chai // build a map of all events in the room // Timeline and State contain different sets of events, so check them both. eventsByID := map[string]*gomatrixserverlib.Event{} + r.TimelineMutex.RLock() for _, ev := range r.Timeline { eventsByID[ev.EventID()] = ev } + r.TimelineMutex.RUnlock() + r.StateMutex.RLock() for _, ev := range r.State { eventsByID[ev.EventID()] = ev } + r.StateMutex.RUnlock() // a queue of events whose auth events are to be included in the auth chain queue := []*gomatrixserverlib.Event{} @@ -156,6 +172,7 @@ func (r *ServerRoom) MustHaveMembershipForUser(t *testing.T, userID, wantMembers func (r *ServerRoom) ServersInRoom() (servers []string) { serverSet := make(map[string]struct{}) + r.StateMutex.RLock() for _, ev := range r.State { if ev.Type() != "m.room.member" { continue @@ -171,6 +188,7 @@ func (r *ServerRoom) ServersInRoom() (servers []string) { serverSet[string(server)] = struct{}{} } + r.StateMutex.RUnlock() for server := range serverSet { servers = append(servers, server) diff --git a/tests/csapi/device_lists_test.go b/tests/csapi/device_lists_test.go index 3a2b0172..bff36fb6 100644 --- a/tests/csapi/device_lists_test.go +++ b/tests/csapi/device_lists_test.go @@ -2,6 +2,7 @@ package csapi_tests import ( "fmt" + "sync/atomic" "testing" "github.com/matrix-org/complement/internal/b" @@ -19,11 +20,11 @@ import ( // 1. `/sync`'s `device_lists.changed/left` contain the correct user IDs. // 2. `/keys/query` returns the correct information after device list updates. func TestDeviceListUpdates(t *testing.T) { - localpartIndex := 0 + var localpartIndex int64 = 0 // generateLocalpart generates a unique localpart based on the given name. generateLocalpart := func(localpart string) string { - localpartIndex++ - return fmt.Sprintf("%s%d", localpart, localpartIndex) + index := atomic.AddInt64(&localpartIndex, 1) + return fmt.Sprintf("%s%d", localpart, index) } // uploadNewKeys uploads a new set of keys for a given client. diff --git a/tests/msc2716_test.go b/tests/msc2716_test.go index 1a69449c..c0e57866 100644 --- a/tests/msc2716_test.go +++ b/tests/msc2716_test.go @@ -14,6 +14,7 @@ import ( "net/http" "net/url" "strings" + "sync/atomic" "testing" "time" @@ -1446,7 +1447,7 @@ func createMessageEventsForBatchSendRequest( "origin_server_ts": insertOriginServerTs + (timeBetweenMessagesMS * uint64(i)), "content": map[string]interface{}{ "msgtype": "m.text", - "body": fmt.Sprintf("Historical %d (batch=%d)", i, batchCount), + "body": fmt.Sprintf("Historical %d (batch=%d)", i, atomic.LoadInt64(&batchCount)), historicalContentField: true, }, } @@ -1514,7 +1515,7 @@ func batchSendHistoricalMessages( t.Fatalf("msc2716.batchSendHistoricalMessages got %d HTTP status code from batch send response but want %d", res.StatusCode, expectedStatus) } - batchCount++ + atomic.AddInt64(&batchCount, 1) return res } diff --git a/tests/room_timestamp_to_event_test.go b/tests/room_timestamp_to_event_test.go index ecace31c..b3085d89 100644 --- a/tests/room_timestamp_to_event_test.go +++ b/tests/room_timestamp_to_event_test.go @@ -11,6 +11,7 @@ import ( "fmt" "net/url" "strconv" + "sync/atomic" "testing" "time" @@ -236,12 +237,12 @@ type eventTime struct { AfterTimestamp time.Time } -var txnCounter int = 0 +var txnCounter int64 = 0 func getTxnID(prefix string) (txnID string) { - txnId := fmt.Sprintf("%s-%d", prefix, txnCounter) + txnId := fmt.Sprintf("%s-%d", prefix, atomic.LoadInt64(&txnCounter)) - txnCounter++ + atomic.AddInt64(&txnCounter, 1) return txnId }