From 06638f0c48d4e45d7c1c77d8d456cd33e3622bd7 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Thu, 18 Jun 2026 19:13:30 +0100 Subject: [PATCH 1/7] Define a type for the port channel ... instead of repeating it 100 times --- internal/deploy/rpc/client.go | 24 +++++++++--------------- 1 file changed, 9 insertions(+), 15 deletions(-) diff --git a/internal/deploy/rpc/client.go b/internal/deploy/rpc/client.go index e2ed338..6d1ef6b 100644 --- a/internal/deploy/rpc/client.go +++ b/internal/deploy/rpc/client.go @@ -63,10 +63,7 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio ct.Fatalf(t, "%s: cannot start RPC binary %s: %s", contextID, r.binaryPath, err) } // wait until we get a high-numbered port - portCh := make(chan struct { - port int - err error - }) + portCh := make(chan portChannelPayload) go func() { rd := bufio.NewReader(stdout) defer close(portCh) @@ -91,10 +88,7 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio str, err := rd.ReadString('\n') if port == 0 { // we need a port if err != nil { - portCh <- struct { - port int - err error - }{port: 0, err: fmt.Errorf("failed to read stdout line: %s", err)} + portCh <- portChannelPayload{port: 0, err: fmt.Errorf("failed to read stdout line: %s", err)} return } port, err = strconv.Atoi(strings.TrimSpace(str)) @@ -102,13 +96,7 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio log.Printf(" RPC (%s) stdout line isn't a port: %s", contextID, str) continue } - portCh <- struct { - port int - err error - }{ - port: port, - err: nil, - } + portCh <- portChannelPayload{port: port, err: nil} break } } @@ -141,6 +129,12 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio panic("unreachable") } +// portChannelPayload is the payload of the channel that we use to communicate the port number from the RPC server to the RPC client. +type portChannelPayload struct { + port int + err error +} + // RPCClient implements api.Client by making RPC calls to an RPC server, which actually has a concrete api.Client type RPCClient struct { client *rpc.Client From da9528b534f650db18a690c1218591f1e0f3cfb3 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Thu, 18 Jun 2026 19:25:06 +0100 Subject: [PATCH 2/7] Factor out a complicated anonymous function to `readPortNumberAndLogOutput` --- internal/deploy/rpc/client.go | 79 +++++++++++++++++++---------------- 1 file changed, 42 insertions(+), 37 deletions(-) diff --git a/internal/deploy/rpc/client.go b/internal/deploy/rpc/client.go index 6d1ef6b..a709b76 100644 --- a/internal/deploy/rpc/client.go +++ b/internal/deploy/rpc/client.go @@ -3,6 +3,7 @@ package rpc import ( "bufio" "fmt" + "io" "log" "net/rpc" "os" @@ -64,43 +65,7 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio } // wait until we get a high-numbered port portCh := make(chan portChannelPayload) - go func() { - rd := bufio.NewReader(stdout) - defer close(portCh) - defer func() { - // log stdout from the RPC server - go func() { - for { - str, err := rd.ReadString('\n') - if err != nil { - log.Print("RPC ERROR: " + err.Error()) - break - } - log.Printf(" RPC (%s): %s", contextID, str) - } - }() - // we need to .Wait to ensure we clean up resources when the RPC server dies. - rpcCmd.Wait() - }() - - var port int - for { - str, err := rd.ReadString('\n') - if port == 0 { // we need a port - if err != nil { - portCh <- portChannelPayload{port: 0, err: fmt.Errorf("failed to read stdout line: %s", err)} - return - } - port, err = strconv.Atoi(strings.TrimSpace(str)) - if err != nil { - log.Printf(" RPC (%s) stdout line isn't a port: %s", contextID, str) - continue - } - portCh <- portChannelPayload{port: port, err: nil} - break - } - } - }() + go readPortNumberAndLogOutput(contextID, portCh, rpcCmd, stdout) select { case p := <-portCh: rpcAddr := fmt.Sprintf("127.0.0.1:%d", p.port) @@ -135,6 +100,46 @@ type portChannelPayload struct { err error } +// readPortNumberAndLogOutput waits for the RPC server to write the port number to stdout, and writes the port number to the `portCh` channel. +// Any other output is logged. +func readPortNumberAndLogOutput(contextID string, portCh chan portChannelPayload, rpcCmd *exec.Cmd, stdout io.ReadCloser) { + rd := bufio.NewReader(stdout) + defer close(portCh) + defer func() { + // log stdout from the RPC server + go func() { + for { + str, err := rd.ReadString('\n') + if err != nil { + log.Print("RPC ERROR: " + err.Error()) + break + } + log.Printf(" RPC (%s): %s", contextID, str) + } + }() + // we need to .Wait to ensure we clean up resources when the RPC server dies. + rpcCmd.Wait() + }() + + var port int + for { + str, err := rd.ReadString('\n') + if port == 0 { // we need a port + if err != nil { + portCh <- portChannelPayload{port: 0, err: fmt.Errorf("failed to read stdout line: %s", err)} + return + } + port, err = strconv.Atoi(strings.TrimSpace(str)) + if err != nil { + log.Printf(" RPC (%s) stdout line isn't a port: %s", contextID, str) + continue + } + portCh <- portChannelPayload{port: port, err: nil} + break + } + } +} + // RPCClient implements api.Client by making RPC calls to an RPC server, which actually has a concrete api.Client type RPCClient struct { client *rpc.Client From 2ff0d68904c3d4e3ea1f5d2c2347199e066e5b9c Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Thu, 18 Jun 2026 19:48:24 +0100 Subject: [PATCH 3/7] RPCClient: clean up logging Make the logging more consistent, and associate it with the relevant test. Also add a few comments about things I don't understand, or which need fixing. --- internal/deploy/rpc/client.go | 33 +++++++++++++++++++++------------ 1 file changed, 21 insertions(+), 12 deletions(-) diff --git a/internal/deploy/rpc/client.go b/internal/deploy/rpc/client.go index a709b76..ceeaa02 100644 --- a/internal/deploy/rpc/client.go +++ b/internal/deploy/rpc/client.go @@ -4,7 +4,6 @@ import ( "bufio" "fmt" "io" - "log" "net/rpc" "os" "os/exec" @@ -54,25 +53,30 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio if _, err := os.Stat(r.binaryPath); err != nil { ct.Fatalf(t, "%s: RPC binary at %s does not exist or cannot be executed/read: %s", contextID, r.binaryPath, err) } + t.Logf("RPC (%s): starting RPC binary at %s", contextID, r.binaryPath) rpcCmd := exec.Command(r.binaryPath) stdout, err := rpcCmd.StdoutPipe() if err != nil { - ct.Fatalf(t, "%s: cannot pipe stdout of rpc binary: %s", contextID, err) + ct.Fatalf(t, "RPC (%s): cannot pipe stdout of rpc binary: %s", contextID, err) } rpcCmd.Stderr = rpcCmd.Stdout if err := rpcCmd.Start(); err != nil { // this calls NewRPCServer() effectively - ct.Fatalf(t, "%s: cannot start RPC binary %s: %s", contextID, r.binaryPath, err) + ct.Fatalf(t, "RPC (%s): cannot start RPC binary %s: %s", contextID, r.binaryPath, err) } // wait until we get a high-numbered port portCh := make(chan portChannelPayload) - go readPortNumberAndLogOutput(contextID, portCh, rpcCmd, stdout) + go readPortNumberAndLogOutput(t, contextID, portCh, rpcCmd, stdout) + + // XXX should we not stop the RPC server once the test is complete, rather than leaving it running to clutter + // up the logs of future tests? + select { case p := <-portCh: rpcAddr := fmt.Sprintf("127.0.0.1:%d", p.port) var void int client, err := rpc.DialHTTP("tcp", rpcAddr) if err != nil { - t.Fatalf("RPC MustCreateClient DialHTTP: %s", err) + ct.Fatalf(t, "RPC (%s): MustCreateClient DialHTTP failed: %s", contextID, err) } err = client.Call("Server.MustCreateClient", ClientCreationOpts{ @@ -81,7 +85,7 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio Lang: r.clientType, }, &void) if err != nil { - ct.Fatalf(t, "%s: failed to create RPC client: %s", contextID, err) + ct.Fatalf(t, "RPC (%s): failed to create RPC client: %s", contextID, err) } return &RPCClient{ client: client, @@ -89,7 +93,7 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio rpcCmd: rpcCmd, } case <-time.After(time.Second): - ct.Fatalf(t, "%s: timed out waiting for port number to be echoed to stdout. Did the RPC binary run, and is it actually the RPC binary? Path: %s", contextID, r.binaryPath) + ct.Fatalf(t, "RPC (%s): timed out waiting for port number to be echoed to stdout. Did the RPC binary run, and is it actually the RPC binary? Path: %s", contextID, r.binaryPath) } panic("unreachable") } @@ -102,19 +106,22 @@ type portChannelPayload struct { // readPortNumberAndLogOutput waits for the RPC server to write the port number to stdout, and writes the port number to the `portCh` channel. // Any other output is logged. -func readPortNumberAndLogOutput(contextID string, portCh chan portChannelPayload, rpcCmd *exec.Cmd, stdout io.ReadCloser) { +func readPortNumberAndLogOutput(t ct.TestLike, contextID string, portCh chan portChannelPayload, rpcCmd *exec.Cmd, stdout io.ReadCloser) { rd := bufio.NewReader(stdout) defer close(portCh) + + // XXX: why is this written as a `defer`, rather than simply happening after the `for` loop, given nobody is + // waiting for `readPortNumberAndLogOutput` to complete? defer func() { // log stdout from the RPC server go func() { for { str, err := rd.ReadString('\n') if err != nil { - log.Print("RPC ERROR: " + err.Error()) + t.Logf("RPC (%s): ERROR: %s", contextID, err) break } - log.Printf(" RPC (%s): %s", contextID, str) + t.Logf("RPC (%s): server output: %s", contextID, str) } }() // we need to .Wait to ensure we clean up resources when the RPC server dies. @@ -124,16 +131,18 @@ func readPortNumberAndLogOutput(contextID string, portCh chan portChannelPayload var port int for { str, err := rd.ReadString('\n') - if port == 0 { // we need a port + if port == 0 { // we need a port. XXX: why would it not be 0, given we `break` having successfully read the port? if err != nil { portCh <- portChannelPayload{port: 0, err: fmt.Errorf("failed to read stdout line: %s", err)} return } port, err = strconv.Atoi(strings.TrimSpace(str)) if err != nil { - log.Printf(" RPC (%s) stdout line isn't a port: %s", contextID, str) + // Not a port number. Log it like a normal line. + t.Logf("RPC (%s): server output: %s", contextID, str) continue } + t.Logf("RPC (%s): server started on port %d", contextID, port) portCh <- portChannelPayload{port: port, err: nil} break } From b361b482102ae25e032ac2b827d9bb1422f9a358 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Fri, 19 Jun 2026 00:30:44 +0100 Subject: [PATCH 4/7] Have the RPC server shut down once someone calls `Server.Close()` --- cmd/rpc/main.go | 18 ++++++++++++++++-- internal/deploy/rpc/server.go | 11 ++++++++++- 2 files changed, 26 insertions(+), 3 deletions(-) diff --git a/cmd/rpc/main.go b/cmd/rpc/main.go index 7bced3f..1b4a654 100644 --- a/cmd/rpc/main.go +++ b/cmd/rpc/main.go @@ -1,6 +1,7 @@ package main import ( + "context" "fmt" "log" "net" @@ -11,7 +12,8 @@ import ( ) func main() { - srv := crpc.NewServer() + srvDoneChannel := make(chan struct{}) + srv := crpc.NewServer(srvDoneChannel) rpc.Register(srv) rpc.HandleHTTP() listener, err := net.Listen("tcp", ":0") @@ -21,5 +23,17 @@ func main() { // tell the parent process what port we are listening on. port := listener.Addr().(*net.TCPAddr).Port fmt.Println(port) - fmt.Println(http.Serve(listener, nil)) + + // Start an HTTP server on the listener. It will send requests to the default HttpMux, which we've + // already told rpc to handle. + httpServer := &http.Server{} + + go httpServer.Serve(listener) + + // Wait for `Server.Close` to be called, then shut down the HTTP server. + <- srvDoneChannel + err = httpServer.Shutdown(context.Background()) + if err != nil { + log.Fatal("HTTP server Shutdown error: ", err) + } } diff --git a/internal/deploy/rpc/server.go b/internal/deploy/rpc/server.go index afe6bc3..59a7eaa 100644 --- a/internal/deploy/rpc/server.go +++ b/internal/deploy/rpc/server.go @@ -28,14 +28,21 @@ type Server struct { waitersMu *sync.Mutex lastCmdRecv time.Time lastCmdRecvMu *sync.Mutex + + // done is a chanel that is closed once Close() has been called. + done chan struct{} } -func NewServer() *Server { +// NewServer creates a new Server instance. +// +// doneChannel is a channel that will be closed once Close() has been called. +func NewServer(doneChannel chan struct{}) *Server { srv := &Server{ waiters: make(map[int]*RPCServerWaiter), waitersMu: &sync.Mutex{}, lastCmdRecv: time.Now(), lastCmdRecvMu: &sync.Mutex{}, + done: doneChannel, } go srv.checkKeepAlive() return srv @@ -91,6 +98,8 @@ func (s *Server) Close(testName string, void *int) error { s.activeClient.Close(&api.MockT{TestName: testName}) // write logs s.bindings.PostTestRun(s.contextID) + // Flag to main that we should shut down the HTTP server, once the client disconnects. + close(s.done) return nil } From b38e555730c209c5a2496eccc9edee341a85ca18 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Fri, 19 Jun 2026 00:45:29 +0100 Subject: [PATCH 5/7] Make RPCClient.Close wait for the logs from the RPC server to be flushed We get a panic if they are written after the test finishes. --- internal/deploy/rpc/client.go | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/internal/deploy/rpc/client.go b/internal/deploy/rpc/client.go index ceeaa02..93bab7d 100644 --- a/internal/deploy/rpc/client.go +++ b/internal/deploy/rpc/client.go @@ -65,10 +65,8 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio } // wait until we get a high-numbered port portCh := make(chan portChannelPayload) - go readPortNumberAndLogOutput(t, contextID, portCh, rpcCmd, stdout) - - // XXX should we not stop the RPC server once the test is complete, rather than leaving it running to clutter - // up the logs of future tests? + logsFlushed := make(chan struct{}) + go readPortNumberAndLogOutput(t, contextID, portCh, rpcCmd, stdout, logsFlushed) select { case p := <-portCh: @@ -91,6 +89,7 @@ func (r *LanguageBindings) MustCreateClient(t ct.TestLike, cfg api.ClientCreatio client: client, lang: r.clientType, rpcCmd: rpcCmd, + logsFlushed: logsFlushed, } case <-time.After(time.Second): ct.Fatalf(t, "RPC (%s): timed out waiting for port number to be echoed to stdout. Did the RPC binary run, and is it actually the RPC binary? Path: %s", contextID, r.binaryPath) @@ -106,7 +105,9 @@ type portChannelPayload struct { // readPortNumberAndLogOutput waits for the RPC server to write the port number to stdout, and writes the port number to the `portCh` channel. // Any other output is logged. -func readPortNumberAndLogOutput(t ct.TestLike, contextID string, portCh chan portChannelPayload, rpcCmd *exec.Cmd, stdout io.ReadCloser) { +// +// Once the RPC server closes and all the logs are flushed, `logsFlushed` is closed. +func readPortNumberAndLogOutput(t ct.TestLike, contextID string, portCh chan portChannelPayload, rpcCmd *exec.Cmd, stdout io.ReadCloser, logsFlushed chan struct{}) { rd := bufio.NewReader(stdout) defer close(portCh) @@ -115,10 +116,11 @@ func readPortNumberAndLogOutput(t ct.TestLike, contextID string, portCh chan por defer func() { // log stdout from the RPC server go func() { + defer close(logsFlushed) for { str, err := rd.ReadString('\n') if err != nil { - t.Logf("RPC (%s): ERROR: %s", contextID, err) + t.Logf("RPC (%s): server closed: %s", contextID, err) break } t.Logf("RPC (%s): server output: %s", contextID, str) @@ -154,6 +156,9 @@ type RPCClient struct { client *rpc.Client lang api.ClientTypeLang rpcCmd *exec.Cmd + + // logsFlushed is a channel that is closed when the log copier goroutine has completed. + logsFlushed chan struct{} } func (c *RPCClient) ForceClose(t ct.TestLike) { @@ -171,12 +176,15 @@ func (c *RPCClient) ForceClose(t ct.TestLike) { func (c *RPCClient) Close(t ct.TestLike) { t.Helper() var void int - fmt.Println("RPCClient.Close") + t.Logf("RPCClient.Close") err := c.client.Call("Server.Close", t.Name(), &void) if err != nil { t.Fatalf("RPCClient.Close: %s", err) } c.client.Close() + + // Wait for the goroutine that copies stdout to the logs to complete + <- c.logsFlushed } func (c *RPCClient) GetNotification(t ct.TestLike, roomID, eventID string) (*api.Notification, error) { From 327c309e75980091c29153b8c281412b7cde7671 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Fri, 19 Jun 2026 00:46:55 +0100 Subject: [PATCH 6/7] Fix internal test to call client.Close() on the clients it creates --- internal/tests/client_test.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/internal/tests/client_test.go b/internal/tests/client_test.go index bf4ed13..5ea847b 100644 --- a/internal/tests/client_test.go +++ b/internal/tests/client_test.go @@ -219,15 +219,20 @@ func TestSendingEvents(t *testing.T) { // run a subtest for each client factory func ForEachClient(t *testing.T, name string, deployment *deploy.ComplementCryptoDeployment, fn func(t *testing.T, client api.TestClient, csapi *client.CSAPI)) { - for _, createClient := range clientFactories { + testWrapper := func (createClient func (t *testing.T, cfg api.ClientCreationOpts) api.TestClient) { csapiAlice := deployment.Register(t, "hs1", helpers.RegistrationOpts{ LocalpartSuffix: "client", Password: "complement-crypto-password", }) opts := api.NewClientCreationOpts(csapiAlice) client := createClient(t, opts) + defer client.Close(t) t.Run(name+" "+string(client.Type()), func(t *testing.T) { fn(t, client, csapiAlice) }) } + + for _, createClient := range clientFactories { + testWrapper(createClient) + } } From d6953ffea7e8deba354bbca868aa73302586efe2 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Fri, 19 Jun 2026 11:19:49 +0100 Subject: [PATCH 7/7] Suppress errors encountered during RPC server shutdown I don't understand why these are happening, but I don't think they are hugely important. Let's just log them and move on. --- cmd/rpc/main.go | 3 +++ internal/deploy/rpc/client.go | 15 +++++++++++---- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/cmd/rpc/main.go b/cmd/rpc/main.go index 1b4a654..e8d40fa 100644 --- a/cmd/rpc/main.go +++ b/cmd/rpc/main.go @@ -32,8 +32,11 @@ func main() { // Wait for `Server.Close` to be called, then shut down the HTTP server. <- srvDoneChannel + fmt.Println("Server starting clean shutdown...") err = httpServer.Shutdown(context.Background()) if err != nil { log.Fatal("HTTP server Shutdown error: ", err) } + + fmt.Println("Server shutdown complete") } diff --git a/internal/deploy/rpc/client.go b/internal/deploy/rpc/client.go index 93bab7d..5830aa5 100644 --- a/internal/deploy/rpc/client.go +++ b/internal/deploy/rpc/client.go @@ -177,14 +177,21 @@ func (c *RPCClient) Close(t ct.TestLike) { t.Helper() var void int t.Logf("RPCClient.Close") - err := c.client.Call("Server.Close", t.Name(), &void) - if err != nil { - t.Fatalf("RPCClient.Close: %s", err) + if err := c.client.Call("Server.Close", t.Name(), &void); err != nil { + // XXX this fails with "unexpected EOF" sometimes, and I don't understand why. + // The server won't actually stop until all clients have disconnected, so the RPC call should + // complete cleanly. + t.Logf("RPCClient.Close: Server.Close RPC call returned error: %s", err) + } + t.Logf("RPCClient.Close: disconnecting RPC client") + if err := c.client.Close(); err != nil { + t.Logf("RPCClient.Close: error disconnecting RPC client: %s", err) } - c.client.Close() // Wait for the goroutine that copies stdout to the logs to complete + t.Logf("RPCClient.Close: waiting for server to shut down") <- c.logsFlushed + t.Logf("RPCClient.Close: done") } func (c *RPCClient) GetNotification(t ct.TestLike, roomID, eventID string) (*api.Notification, error) {