chore: support sharding/parallel runs, namespace scopes via COMPLEMENT_CRYPTO_NAMESPACE

.gitignore

@@ -15,3 +15,5 @@ __pycache__
 /tests/chromedp/
 /tests/rust_storage/
 _temp_rust_sdk
+# Locally-built cmd/rpc binary (COMPLEMENT_CRYPTO_RPC_BINARY target) - machine/toolchain-specific.
+/rpc

cmd/rpc/main.go

@@ -31,7 +31,7 @@ func main() {
 	go httpServer.Serve(listener)
 
 	// Wait for `Server.Close` to be called, then shut down the HTTP server.
-	<- srvDoneChannel
+	<-srvDoneChannel
 	fmt.Println("Server starting clean shutdown...")
 	err = httpServer.Shutdown(context.Background())
 	if err != nil {

internal/api/client.go

@@ -52,7 +52,12 @@ type Client interface {
 	// SendMessage sends the given text as an encrypted/unencrypted message in the room, depending
 	// if the room is encrypted or not. Returns the event ID of the sent event, so MUST BLOCK until the event has been sent.
 	// If the event cannot be sent, returns an error.
-	SendMessage(t ct.TestLike, roomID, text string) (eventID string, err error)
+	//
+	// An optional timeout may be passed to override the implementation's default wait-for-local-echo
+	// timeout (only the first value is used). This exists for stress tests with unusually large
+	// fan-out (e.g. sending to a room with 100+ recipients, where session establishment can
+	// legitimately take longer than the default) - ordinary call sites should omit it.
+	SendMessage(t ct.TestLike, roomID, text string, timeout ...time.Duration) (eventID string, err error)
 	// Wait until an event is seen in the given room. The checker functions can be custom or you can use
 	// a pre-defined one like api.CheckEventHasMembership, api.CheckEventHasBody, or api.CheckEventHasEventID.
 	WaitUntilEventInRoom(t ct.TestLike, roomID string, checker func(e Event) bool) Waiter
@@ -228,10 +233,10 @@ func (c *LoggedClient) IsRoomEncrypted(t ct.TestLike, roomID string) (bool, erro
 	return c.Client.IsRoomEncrypted(t, roomID)
 }
 
-func (c *LoggedClient) SendMessage(t ct.TestLike, roomID, text string) (eventID string, err error) {
+func (c *LoggedClient) SendMessage(t ct.TestLike, roomID, text string, timeout ...time.Duration) (eventID string, err error) {
 	t.Helper()
 	c.Logf(t, "%s SendMessage %s => %s", c.logPrefix(), roomID, text)
-	eventID, err = c.Client.SendMessage(t, roomID, text)
+	eventID, err = c.Client.SendMessage(t, roomID, text, timeout...)
 	c.Logf(t, "%s SendMessage %s => %s %s", c.logPrefix(), roomID, eventID, err)
 	return
 }
@@ -254,7 +259,7 @@ func (c *LoggedClient) Backpaginate(t ct.TestLike, roomID string, count int) err
 		result = fmt.Sprintf("ERROR: %v", err)
 	}
 	c.Logf(t, "%s Backpaginate %d %s => %s", c.logPrefix(), count, roomID, result)
-	
+
 	return err
 }
 

internal/api/js/js-sdk/package.json

@@ -10,7 +10,7 @@
   "license": "Apache-2.0",
   "dependencies": {
     "buffer": "^6.0.3",
-    "matrix-js-sdk": "^41.0.0",
+    "matrix-js-sdk": "https://github.com/matrix-org/matrix-js-sdk#develop",
     "vite": "^6.4.2"
   },
   "packageManager": "yarn@1.22.22"

internal/api/js/js-sdk/yarn.lock

@@ -2,10 +2,10 @@
 # yarn lockfile v1
 
 
-"@babel/runtime@^7.12.5":
-  version "7.28.4"
-  resolved "https://registry.yarnpkg.com/@babel/runtime/-/runtime-7.28.4.tgz#a70226016fabe25c5783b2f22d3e1c9bc5ca3326"
-  integrity sha512-Q/N6JNWvIvPnLDvjlE1OUBLPQHH6l3CltCEsHIujp45zQUSSh8K+gHnaEX45yAT1nyngnINhvWtzN+Nb9D8RAQ==
+"@babel/runtime@^8.0.0":
+  version "8.0.0"
+  resolved "https://registry.yarnpkg.com/@babel/runtime/-/runtime-8.0.0.tgz#d7bd513e6843662346552c2798ab895716cf97f2"
+  integrity sha512-sL6cvO2IfkSu/iU+zs2S/w01B7A8V7suXSIKEN4hPFFdZoiPGxrj5pAG0lCaqLWiEIrjKzdznIWuaLcxPR53qw==
 
 "@esbuild/aix-ppc64@0.25.12":
   version "0.25.12"
@@ -137,10 +137,10 @@
   resolved "https://registry.yarnpkg.com/@esbuild/win32-x64/-/win32-x64-0.25.12.tgz#9bdad8176be7811ad148d1f8772359041f46c6c5"
   integrity sha512-alJC0uCZpTFrSL0CCDjcgleBXPnCrEAhTBILpeAp7M/OFgoqtAetfBzX0xM00MUsVVPpVjlPuMbREqnZCXaTnA==
 
-"@matrix-org/matrix-sdk-crypto-wasm@^18.2.0":
-  version "18.3.0"
-  resolved "https://registry.yarnpkg.com/@matrix-org/matrix-sdk-crypto-wasm/-/matrix-sdk-crypto-wasm-18.3.0.tgz#a5206792a78a3ebebd901662c65753e0914d204a"
-  integrity sha512-9a4feyt8QLysARu7PHKaRWT+wcCd+IYH074LXp9QK5WqfN4zUXueRhiSSMNT18Bm+8q3sBR/4zxDxOSDR0M8Kg==
+"@matrix-org/matrix-sdk-crypto-wasm@^18.4.0":
+  version "18.7.0"
+  resolved "https://registry.yarnpkg.com/@matrix-org/matrix-sdk-crypto-wasm/-/matrix-sdk-crypto-wasm-18.7.0.tgz#81cf44109b91439a57c500e960c6dac03ea77835"
+  integrity sha512-IiO8YrahwBN23iii4OBS6AUIEDZePa72bK7WN/DBjOXPc6g/wt891vn0vxmlopTahqFFOz/LErcHisdmzKVTsg==
 
 "@rollup/rollup-android-arm-eabi@4.60.4":
   version "4.60.4"
@@ -307,10 +307,10 @@ buffer@^6.0.3:
     base64-js "^1.3.1"
     ieee754 "^1.2.1"
 
-content-type@^1.0.4:
-  version "1.0.5"
-  resolved "https://registry.yarnpkg.com/content-type/-/content-type-1.0.5.tgz#8b773162656d1d1086784c8f23a54ce6d73d7918"
-  integrity sha512-nTjqfcBFEipKdXCv4YDQWCfmcLZKm81ldF0pAopTvyrFGVbcR6P/VAAd5G7N+0tTr8QqiU0tFadD6FK4NtJwOA==
+content-type@^3.0.0:
+  version "3.0.0"
+  resolved "https://registry.yarnpkg.com/content-type/-/content-type-3.0.0.tgz#c946981e09aba276f5f5b21223bd0c01fded9c15"
+  integrity sha512-AIi5H6p0xk5uknXcN3/rmhP8jgp69OfSe/JuKiQAFprJ7UGw7mwj7m4XcmDzlrnJDG+cGpphAINGdU3g3g7kDw==
 
 esbuild@^0.25.0:
   version "0.25.12"
@@ -369,11 +369,6 @@ is-network-error@^1.3.0:
   resolved "https://registry.yarnpkg.com/is-network-error/-/is-network-error-1.3.2.tgz#9460bc30f8419a4bca77114f4de88a3ee5e0c519"
   integrity sha512-PhBY86zaxNZUuWP6h13Vu5oFe0XY6/UlKzQnYFELzGVHygP3MxmvTfYSG7GN3aIab/iWudSMgjSnG9Dq+nHrgA==
 
-jwt-decode@^4.0.0:
-  version "4.0.0"
-  resolved "https://registry.yarnpkg.com/jwt-decode/-/jwt-decode-4.0.0.tgz#2270352425fd413785b2faf11f6e755c5151bd4b"
-  integrity sha512-+KJGIyHgkGuIq3IEBNftfhW/LfWhXUIY6OmyVWjliu5KH1y0fw7VQ8YndE2O4qZdMSd9SqbnC8GOcZEy0Om7sA==
-
 loglevel@^1.9.2:
   version "1.9.2"
   resolved "https://registry.yarnpkg.com/loglevel/-/loglevel-1.9.2.tgz#c2e028d6c757720107df4e64508530db6621ba08"
@@ -384,29 +379,26 @@ matrix-events-sdk@0.0.1:
   resolved "https://registry.yarnpkg.com/matrix-events-sdk/-/matrix-events-sdk-0.0.1.tgz#c8c38911e2cb29023b0bbac8d6f32e0de2c957dd"
   integrity sha512-1QEOsXO+bhyCroIe2/A5OwaxHvBm7EsSQ46DEDn8RBIfQwN5HWBpFvyWWR4QY0KHPPnnJdI99wgRiAl7Ad5qaA==
 
-matrix-js-sdk@^41.0.0:
-  version "41.5.0"
-  resolved "https://registry.yarnpkg.com/matrix-js-sdk/-/matrix-js-sdk-41.5.0.tgz#8519fc9a4626be7fe3f6ebe1ed13019e0b097314"
-  integrity sha512-CK3h+qQJ4wkVEUgEWc5MdLjccXyiFqncCC53P+auqOhnX2U6tAFsRfnbML1QQiKIsFMzqTrAnF/4a5LUUOIeXg==
+"matrix-js-sdk@https://github.com/matrix-org/matrix-js-sdk#develop":
+  version "42.2.0"
+  resolved "https://github.com/matrix-org/matrix-js-sdk#aa1aeed63b5ef0327f0442d63e31efca58df46b0"
   dependencies:
-    "@babel/runtime" "^7.12.5"
-    "@matrix-org/matrix-sdk-crypto-wasm" "^18.2.0"
+    "@babel/runtime" "^8.0.0"
+    "@matrix-org/matrix-sdk-crypto-wasm" "^18.4.0"
     another-json "^0.2.0"
     bs58 "^6.0.0"
-    content-type "^1.0.4"
-    jwt-decode "^4.0.0"
+    content-type "^3.0.0"
     loglevel "^1.9.2"
     matrix-events-sdk "0.0.1"
-    matrix-widget-api "^1.16.1"
-    oidc-client-ts "^3.0.1"
+    matrix-widget-api "^1.18.0"
     p-retry "8"
     sdp-transform "^3.0.0"
     unhomoglyph "^1.0.6"
 
-matrix-widget-api@^1.16.1:
-  version "1.17.0"
-  resolved "https://registry.yarnpkg.com/matrix-widget-api/-/matrix-widget-api-1.17.0.tgz#2336de2186fe70d8bd741c1603c162f60b2099c2"
-  integrity sha512-5FHoo3iEP3Bdlv5jsYPWOqj+pGdFQNLWnJLiB0V7Ygne7bb+Gsj3ibyFyHWC6BVw+Z+tSW4ljHpO17I9TwStwQ==
+matrix-widget-api@^1.18.0:
+  version "1.19.0"
+  resolved "https://registry.yarnpkg.com/matrix-widget-api/-/matrix-widget-api-1.19.0.tgz#77819e0307adbba972ae8178c83b77fada896e29"
+  integrity sha512-Qe7L6G1mQJphBC47/Bzp5ljPIK1UARgLiuGgS+/KmEs/eaNHkQ3+ArD0/0kvMV+d6kCGAZ9OKRqsnuPjizI+tw==
   dependencies:
     "@types/events" "^3.0.0"
     events "^3.2.0"
@@ -416,13 +408,6 @@ nanoid@^3.3.11:
   resolved "https://registry.yarnpkg.com/nanoid/-/nanoid-3.3.11.tgz#4f4f112cefbe303202f2199838128936266d185b"
   integrity sha512-N8SpfPUnUp1bK+PMYW8qSWdl9U+wwNWI4QKxOYDy9JAro3WMX7p2OeVRF9v+347pnakNevPmiHhNmZ2HbFA76w==
 
-oidc-client-ts@^3.0.1:
-  version "3.4.1"
-  resolved "https://registry.yarnpkg.com/oidc-client-ts/-/oidc-client-ts-3.4.1.tgz#7cad95ba7213cb93b7c141965ce03ad658baa30f"
-  integrity sha512-jNdst/U28Iasukx/L5MP6b274Vr7ftQs6qAhPBCvz6Wt5rPCA+Q/tUmCzfCHHWweWw5szeMy2Gfrm1rITwUKrw==
-  dependencies:
-    jwt-decode "^4.0.0"
-
 p-retry@8:
   version "8.0.0"
   resolved "https://registry.yarnpkg.com/p-retry/-/p-retry-8.0.0.tgz#1505faf14942326e18d4091f5d605bae65634e7b"

internal/api/js/js.go

@@ -518,10 +518,12 @@ func (c *JSClient) GetEvent(t ct.TestLike, roomID, eventID string) (*api.Event,
 	// }
 	// else just returns { event }
 	evSerialised, err := chrome.RunAsyncFn[string](t, c.browser.Ctx, fmt.Sprintf(`
-	return JSON.stringify(window.__client.getRoom("%s")?.getLiveTimeline()?.getEvents().filter((ev, i) => {
-		console.log("MustGetEvent["+i+"] => " + ev.getId()+ " " + JSON.stringify(ev.toJSON()));
-		return ev.getId() === "%s";
-	})[0].toJSON());
+	const room = window.__client.getRoom("%s");
+	const ev = room?.findEventById("%s");
+	if (!ev) {
+		throw new Error("event not found in room timelines");
+	}
+	return JSON.stringify(ev.toJSON());
 	`, roomID, eventID))
 	if err != nil {
 		return nil, fmt.Errorf("failed to get event %s: %s", eventID, err)
@@ -561,9 +563,11 @@ func (c *JSClient) GetEventShield(t ct.TestLike, roomID, eventID string) (*api.E
 	//    shieldReason: 0 ... 7
 	// }
 	encryptionInfoSerialised, err := chrome.RunAsyncFn[string](t, c.browser.Ctx, fmt.Sprintf(`
-		const ev = window.__client.getRoom("%s")?.getLiveTimeline()?.getEvents().filter((ev, i) => {
-			return ev.getId() === "%s";
-		})[0];
+		const room = window.__client.getRoom("%s");
+		const ev = room?.findEventById("%s");
+		if (!ev) {
+			throw new Error("event not found in room timelines");
+		}
 		const encryptionInfo = await window.__client.getCrypto().getEncryptionInfoForEvent(ev);
 		return JSON.stringify(encryptionInfo);
 	`, roomID, eventID))
@@ -688,8 +692,11 @@ func (c *JSClient) IsRoomEncrypted(t ct.TestLike, roomID string) (bool, error) {
 	return *isEncrypted, nil
 }
 
-func (c *JSClient) SendMessage(t ct.TestLike, roomID, text string) (eventID string, err error) {
+func (c *JSClient) SendMessage(t ct.TestLike, roomID, text string, timeout ...time.Duration) (eventID string, err error) {
 	t.Helper()
+	// JS has no internal wait-for-local-echo timeout to override (chrome.RunAsyncFn awaits
+	// the underlying JS promise directly), so a caller-supplied timeout is a no-op here.
+	_ = timeout
 	res, err := chrome.RunAsyncFn[map[string]interface{}](t, c.browser.Ctx, fmt.Sprintf(`
 	return await window.__client.sendMessage("%s", {
 		"msgtype": "m.text",

internal/api/rust/rust.go

@@ -23,7 +23,7 @@ import (
 
 // LogTarget is the name of the `target` we use in logToFile: it is in effect a fake "crate" that we tell the
 // rust-sdk is producing the logs.
-const LogTarget = "complement_crypto";
+const LogTarget = "complement_crypto"
 
 func DeleteOldLogs(prefix string) {
 	// delete old log files
@@ -39,7 +39,7 @@ func SetupLogs(prefix string) {
 	// log new files
 	matrix_sdk_ffi.InitPlatform(matrix_sdk_ffi.TracingConfiguration{
 		LogLevel:              matrix_sdk_ffi.LogLevelTrace,
-		ExtraTargets:          []string {LogTarget},
+		ExtraTargets:          []string{LogTarget},
 		WriteToStdoutOrSystem: false,
 		WriteToFiles: &matrix_sdk_ffi.TracingFileConfiguration{
 			Path:       "./logs",
@@ -104,11 +104,10 @@ func NewRustClient(t ct.TestLike, opts api.ClientCreationOpts) (api.Client, erro
 	xprocessName := opts.GetExtraOption(CrossProcessStoreLocksHolderName, "").(string)
 	if xprocessName != "" {
 		t.Logf("setting cross process store locks holder name=%s", xprocessName)
-		ab = ab.CrossProcessLockConfig(matrix_sdk_ffi.CrossProcessLockConfigMultiProcess { xprocessName })
+		ab = ab.CrossProcessLockConfig(matrix_sdk_ffi.CrossProcessLockConfigMultiProcess{xprocessName})
 	}
 	// @alice:hs1, FOOBAR => alice_hs1_FOOBAR
 	username := strings.Replace(opts.UserID[1:], ":", "_", -1) + "_" + opts.DeviceID
-	ab = ab.Username(username)
 
 	sessionPath := "rust_storage/" + username
 	storeKey := []byte("my_secret_thirty-two_byte_string")
@@ -675,11 +674,15 @@ func (c *RustClient) Type() api.ClientTypeLang {
 	return api.ClientTypeRust
 }
 
-func (c *RustClient) SendMessage(t ct.TestLike, roomID, text string) (eventID string, err error) {
+func (c *RustClient) SendMessage(t ct.TestLike, roomID, text string, timeout ...time.Duration) (eventID string, err error) {
 	c.FFISpan.Enter()
 	defer c.FFISpan.Exit()
 
 	t.Helper()
+	waitFor := 11 * time.Second
+	if len(timeout) > 0 {
+		waitFor = timeout[0]
+	}
 	var isChannelClosed atomic.Bool
 	ch := make(chan bool)
 	// we need a timeline listener before we can send messages, AND that listener must be attached to the
@@ -723,8 +726,8 @@ func (c *RustClient) SendMessage(t ct.TestLike, roomID, text string) (eventID st
 	}
 	timeline.Send(matrix_sdk_ffi.MessageEventContentFromHtml(text, text))
 	select {
-	case <-time.After(11 * time.Second):
-		err = fmt.Errorf("SendMessage(rust) %s: timed out after 11s", c.userID)
+	case <-time.After(waitFor):
+		err = fmt.Errorf("SendMessage(rust) %s: timed out after %s", c.userID, waitFor)
 		return
 	case <-ch:
 		return
@@ -829,7 +832,7 @@ func (c *RustClient) Logf(t ct.TestLike, format string, args ...interface{}) {
 func (c *RustClient) logToFile(t ct.TestLike, format string, args ...interface{}) {
 	c.FFISpan.Enter()
 	defer c.FFISpan.Exit()
-	matrix_sdk_ffi.LogEvent("rust.go", &zero, matrix_sdk_ffi.LogLevelInfo, LogTarget + "::" + t.Name(), fmt.Sprintf(format, args...))
+	matrix_sdk_ffi.LogEvent("rust.go", &zero, matrix_sdk_ffi.LogLevelInfo, LogTarget+"::"+t.Name(), fmt.Sprintf(format, args...))
 }
 
 func (c *RustClient) ensureListening(t ct.TestLike, roomID string) {
@@ -863,6 +866,22 @@ func (c *RustClient) ensureListening(t ct.TestLike, roomID string) {
 		return
 	}
 
+	// Without an explicit room subscription, the sliding sync `pos` for this room
+	// only ever advances via whatever small timeline_limit the "all rooms" list
+	// uses for previews (e.g. 1-10). That's fine when events trickle in one at a
+	// time, but under concurrent load a burst of room state (joins, membership
+	// changes) plus messages can exceed that window in a single poll; the server
+	// correctly reports `limited: true` with a `prev_batch` pointing before the
+	// gap, but nothing here ever triggers backpagination to close it, so a
+	// message that fell into the truncated portion is silently never delivered.
+	// Subscribing before we start consuming the timeline requests a much larger
+	// window (SDK default: 20) on every subsequent poll for this room, giving
+	// real headroom so this doesn't happen in the first place - this mirrors
+	// what a real client does when a room is actually open/visible.
+	if err := c.syncService.RoomListService().SubscribeToRooms([]string{roomID}); err != nil {
+		c.Logf(t, "[%s]ensureListening[%s] failed to subscribe to room: %s", c.userID, roomID, err)
+	}
+
 	c.Logf(t, "[%s]AddTimelineListener[%s]", c.userID, roomID)
 	// we need a timeline listener before we can send messages. Ensure we insert the initial
 	// set of items prior to handling updates. If we don't wait, we risk the listener firing
@@ -936,6 +955,22 @@ func (c *RustClient) ensureListening(t ct.TestLike, roomID string) {
 				ev := timelineItemToEvent(x.Value)
 				timeline = slices.Insert(timeline, 0, ev)
 				newEvents = append(newEvents, ev)
+			case matrix_sdk_ffi.TimelineDiffClear:
+				timeline = make([]*api.Event, 0)
+				c.logToFile(t, "[%s]_______ CLEAR", c.userID)
+			case matrix_sdk_ffi.TimelineDiffPopFront:
+				if len(timeline) > 0 {
+					timeline = slices.Delete(timeline, 0, 1)
+				}
+			case matrix_sdk_ffi.TimelineDiffPopBack:
+				if len(timeline) > 0 {
+					timeline = slices.Delete(timeline, len(timeline)-1, len(timeline))
+				}
+			case matrix_sdk_ffi.TimelineDiffTruncate:
+				n := int(x.Length)
+				if n < len(timeline) {
+					timeline = timeline[:n]
+				}
 			default:
 				t.Logf("Unhandled TimelineDiff change %v", d)
 			}

internal/deploy/rpc/client.go

@@ -86,9 +86,9 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio
 			ct.Fatalf(t, "RPC (%s): failed to create RPC client: %s", contextID, err)
 		}
 		return &RPCClient{
-			client: client,
-			lang:   r.clientType,
-			rpcCmd: rpcCmd,
+			client:      client,
+			lang:        r.clientType,
+			rpcCmd:      rpcCmd,
 			logsFlushed: logsFlushed,
 		}
 	case <-time.After(time.Second):
@@ -190,7 +190,7 @@ func (c *RPCClient) Close(t ct.TestLike) {
 
 	// Wait for the goroutine that copies stdout to the logs to complete
 	t.Logf("RPCClient.Close: waiting for server to shut down")
-	<- c.logsFlushed
+	<-c.logsFlushed
 	t.Logf("RPCClient.Close: done")
 }
 
@@ -277,11 +277,16 @@ func (c *RPCClient) IsRoomEncrypted(t ct.TestLike, roomID string) (bool, error)
 }
 
 // SendMessage tries to send the message, but can fail.
-func (c *RPCClient) SendMessage(t ct.TestLike, roomID, text string) (eventID string, err error) {
+func (c *RPCClient) SendMessage(t ct.TestLike, roomID, text string, timeout ...time.Duration) (eventID string, err error) {
+	var to time.Duration
+	if len(timeout) > 0 {
+		to = timeout[0]
+	}
 	err = c.client.Call("Server.SendMessage", RPCSendMessage{
 		TestName: t.Name(),
 		RoomID:   roomID,
 		Text:     text,
+		Timeout:  to,
 	}, &eventID)
 	return
 }