test(e2e): add real multi-client concurrent reproducer for issue #95

internal/e2e/multiclient_real_test.go

@@ -0,0 +1,181 @@
+package e2e
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/openlibrecommunity/olcrtc/internal/app/session"
+	"github.com/openlibrecommunity/olcrtc/internal/client"
+	"github.com/openlibrecommunity/olcrtc/internal/control"
+	"github.com/openlibrecommunity/olcrtc/internal/server"
+)
+
+// TestRealMultiClientConcurrent is the reproduction harness for issue #95's
+// real-world symptom: multiple clients sharing one server in a single SFU
+// room. The single-client matrix (TestRealProviderTransportMatrix) passes, so
+// any failure here isolates the multi-peer control-plane defect rather than a
+// generic connectivity problem.
+//
+// Topology: one server + two clients in the same Telemost room, each with its
+// own device id and SOCKS listener. We verify:
+//
+//  1. Both clients reach READY (handshake completes for BOTH, not just the
+//     first - the log showed client 2/3 hanging in "read welcome: timeout").
+//  2. Both can echo concurrently right after connect.
+//  3. After holding the link past one liveness interval (the log showed
+//     teardown ~10-15s in via "control missed pong"), both clients STILL
+//     echo - i.e. the session was not torn down underneath a healthy link.
+//
+// Gated behind -olcrtc.real-e2e so it never runs on a normal push.
+func TestRealMultiClientConcurrent(t *testing.T) {
+	if !*realE2E {
+		t.Skip("real provider e2e disabled; pass -olcrtc.real-e2e with provider room flags")
+	}
+
+	const (
+		carrierName   = "telemost"
+		transportName = "vp8channel"
+		holdDuration  = 25 * time.Second
+	)
+
+	echoAddr := startEchoServer(t)
+
+	roomCtx, cancelRoom := context.WithTimeout(context.Background(), *realE2ETimeout)
+	roomURL := requireRealRoom(roomCtx, t, carrierName)
+	cancelRoom()
+
+	ctx, cancel := context.WithTimeout(context.Background(), holdDuration+3*(*realE2ETimeout))
+	defer cancel()
+
+	session.RegisterDefaults()
+
+	// Shared channel id: server and both clients must bind to the same room
+	// instance so the server's peer-routing demux sees both client epochs.
+	channelID := fmt.Sprintf("e2e-multi-%d", time.Now().UnixNano())
+	liveness := control.Config{Interval: 10 * time.Second, Timeout: 60 * time.Second, Failures: 10}
+
+	serverErr := make(chan error, 1)
+	go func() {
+		serverErr <- server.Run(ctx, server.Config{
+			Transport:        transportName,
+			Carrier:          carrierName,
+			RoomURL:          roomURL,
+			ChannelID:        channelID,
+			KeyHex:           testKeyHex,
+			DNSServer:        localDNSServer,
+			TransportOptions: e2eTransportOptions(transportName),
+			Liveness:         liveness,
+		})
+	}()
+
+	select {
+	case err := <-serverErr:
+		t.Fatalf("server exited before clients started: %v", err)
+	case <-time.After(2 * time.Second):
+	}
+
+	clientA := startNamedClient(ctx, t, clientSpec{
+		carrier: carrierName, transport: transportName, room: roomURL,
+		channelID: channelID, deviceID: "client-A", liveness: liveness,
+	})
+	clientB := startNamedClient(ctx, t, clientSpec{
+		carrier: carrierName, transport: transportName, room: roomURL,
+		channelID: channelID, deviceID: "client-B", liveness: liveness,
+	})
+
+	// 1. Both clients must reach READY.
+	waitForReadyWithin(t, clientA.ready, *realE2ETimeout)
+	t.Log("client-A ready")
+	waitForReadyWithin(t, clientB.ready, *realE2ETimeout)
+	t.Log("client-B ready")
+
+	// 2. Both echo right after connect.
+	mustEcho(t, clientA.socksAddr, echoAddr, "A-initial")
+	mustEcho(t, clientB.socksAddr, echoAddr, "B-initial")
+	t.Log("both clients echoed immediately after connect")
+
+	// 3. Hold past the liveness window, then both must still echo.
+	select {
+	case err := <-serverErr:
+		t.Fatalf("server died during hold: %v", err)
+	case err := <-clientA.errc:
+		t.Fatalf("client-A died during hold: %v", err)
+	case err := <-clientB.errc:
+		t.Fatalf("client-B died during hold: %v", err)
+	case <-time.After(holdDuration):
+	}
+
+	mustEcho(t, clientA.socksAddr, echoAddr, "A-after-hold")
+	mustEcho(t, clientB.socksAddr, echoAddr, "B-after-hold")
+	t.Logf("both clients still alive and echoing after %s hold", holdDuration)
+
+	cancel()
+}
+
+type clientSpec struct {
+	carrier   string
+	transport string
+	room      string
+	channelID string
+	deviceID  string
+	liveness  control.Config
+}
+
+type namedClient struct {
+	socksAddr string
+	ready     chan struct{}
+	errc      chan error
+}
+
+func startNamedClient(ctx context.Context, t *testing.T, spec clientSpec) namedClient {
+	t.Helper()
+	socksAddr := freeLocalAddr(ctx, t)
+	ready := make(chan struct{})
+	errc := make(chan error, 1)
+	go func() {
+		errc <- client.RunWithReady(ctx, client.Config{
+			Transport:        spec.transport,
+			Carrier:          spec.carrier,
+			RoomURL:          spec.room,
+			ChannelID:        spec.channelID,
+			KeyHex:           testKeyHex,
+			DeviceID:         spec.deviceID,
+			LocalAddr:        socksAddr,
+			DNSServer:        localDNSServer,
+			TransportOptions: e2eTransportOptions(spec.transport),
+			Liveness:         spec.liveness,
+		}, func() { close(ready) })
+	}()
+	return namedClient{socksAddr: socksAddr, ready: ready, errc: errc}
+}
+
+func mustEcho(t *testing.T, socksAddr, echoAddr, tag string) {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), *realE2ETimeout)
+	defer cancel()
+
+	conn, err := connectViaSOCKSWithin(ctx, socksAddr, echoAddr, *realE2ETimeout)
+	if err != nil {
+		t.Fatalf("[%s] socks connect: %v", tag, err)
+	}
+	defer func() { _ = conn.Close() }()
+
+	payload := []byte("olcrtc-multi-" + tag + "\n")
+	if _, err := conn.Write(payload); err != nil {
+		t.Fatalf("[%s] write: %v", tag, err)
+	}
+	if err := conn.SetReadDeadline(time.Now().Add(*realE2ETimeout)); err != nil {
+		t.Fatalf("[%s] set deadline: %v", tag, err)
+	}
+	line, err := bufio.NewReader(conn).ReadBytes('\n')
+	if err != nil {
+		t.Fatalf("[%s] read echo: %v", tag, err)
+	}
+	if !bytes.Equal(line, payload) {
+		t.Fatalf("[%s] echo mismatch: got %q want %q", tag, line, payload)
+	}
+}