Increase thread safety on variables

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{}{

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

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)
 

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)

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.

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
 }

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
 }