mqtt-5-test/internal/integration/reconnect_test.go

package integration

import (
	"context"
	"io"
	"net"
	"strings"
	"sync"
	"testing"
	"time"

	"robot-market/test-mqtt-5/internal/backend"
	"robot-market/test-mqtt-5/internal/events"
	"robot-market/test-mqtt-5/internal/mqtt"
	"robot-market/test-mqtt-5/internal/simulator"
)

// A loopback byte relay drops only this test client's connection. No broker changes.
type relay struct {
	listener    net.Listener
	target      string
	mu          sync.Mutex
	paused      bool
	connections map[net.Conn]bool
	wg          sync.WaitGroup
}

func newRelay(t *testing.T, target string) *relay {
	t.Helper()
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	r := &relay{listener: listener, target: target, connections: make(map[net.Conn]bool)}
	r.wg.Add(1)
	go func() {
		defer r.wg.Done()
		for {
			local, err := listener.Accept()
			if err != nil {
				return
			}
			r.mu.Lock()
			if r.paused {
				r.mu.Unlock()
				local.Close()
				continue
			}
			r.connections[local] = true
			r.wg.Add(1)
			r.mu.Unlock()
			go r.forward(local)
		}
	}()
	return r
}
func (r *relay) forward(local net.Conn) {
	defer r.wg.Done()
	defer local.Close()
	defer func() { r.mu.Lock(); delete(r.connections, local); r.mu.Unlock() }()
	remote, err := net.DialTimeout("tcp", r.target, 3*time.Second)
	if err != nil {
		return
	}
	defer remote.Close()
	r.mu.Lock()
	if r.paused {
		r.mu.Unlock()
		return
	}
	r.connections[remote] = true
	r.mu.Unlock()
	defer func() { r.mu.Lock(); delete(r.connections, remote); r.mu.Unlock() }()
	done := make(chan struct{})
	go func() { io.Copy(remote, local); remote.Close(); close(done) }()
	io.Copy(local, remote)
	local.Close()
	<-done
}
func (r *relay) pause() {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.paused = true
	for c := range r.connections {
		c.Close()
	}
}
func (r *relay) resume() { r.mu.Lock(); r.paused = false; r.mu.Unlock() }
func (r *relay) close()  { r.listener.Close(); r.pause(); r.wg.Wait() }
func TestRemoteReconnectSessionAndUnknown(t *testing.T) {
	for _, resumed := range []bool{true, false} {
		name := "SessionResumed"
		if !resumed {
			name = "SessionLostSubscriptionsRestored"
		}
		t.Run(name, func(t *testing.T) { testReconnect(t, resumed) })
	}
}
func TestRemoteSessionExpiry(t *testing.T) {
	testReconnectExpiry(t, false, time.Second)
}
func testReconnect(t *testing.T, expectResumed bool) { testReconnectExpiry(t, expectResumed, 0) }
func testReconnectExpiry(t *testing.T, expectResumed bool, expiry time.Duration) {
	cfg := remoteConfig(t, "test-backend", false)
	if !expectResumed {
		cfg.MQTT.SessionExpiry = expiry
	}
	target := strings.TrimPrefix(cfg.MQTT.URL, "tls://")
	proxy := newRelay(t, target)
	defer proxy.close()
	// The generated server certificate explicitly includes 127.0.0.1 for health checks.
	cfg.MQTT.URL = "tls://" + proxy.listener.Addr().String()
	cfg.MQTT.ReconnectMin = 100 * time.Millisecond
	cfg.MQTT.ReconnectMax = 300 * time.Millisecond
	client := mqtt.New(cfg, store(cfg))
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	if err := client.Start(ctx, []string{"acks", "telemetry"}, nil); err != nil {
		t.Fatal(err)
	}
	defer client.Close()
	b := backend.New(client)
	workerCtx, stop := context.WithCancel(ctx)
	workerDone := make(chan error, 1)
	go func() { workerDone <- b.Serve(workerCtx) }()
	defer func() { stop(); <-workerDone }()
	simCfg := remoteConfig(t, "sim-001", false)
	simCfg.Simulator.AckDelay = 500 * time.Millisecond
	simClient := mqtt.New(simCfg, store(simCfg))
	sim := simulator.New(simClient)
	simCtx, simStop := context.WithCancel(ctx)
	simDone := make(chan error, 1)
	go func() { simDone <- sim.Run(simCtx) }()
	defer func() { simStop(); <-simDone }()
	wait(t, simClient.Ready)
	cmdDone := make(chan error, 1)
	go func() {
		_, err := b.Command(ctx, backend.NewCommand("sim-001", "ping", backend.ID(), 5*time.Second))
		cmdDone <- err
	}()
	wait(t, func() bool { _, counts := client.Events.Snapshot(); return counts[events.CommandAccepted] > 0 })
	proxy.pause()
	wait(t, func() bool { return !client.Ready() })
	select {
	case err := <-cmdDone:
		if err == nil || !strings.Contains(err.Error(), "unknown") {
			t.Fatal("connection loss did not produce unknown outcome")
		}
	case <-time.After(3 * time.Second):
		t.Fatal("pending request did not stop on connection loss")
	}
	if expiry > 0 {
		time.Sleep(expiry + 500*time.Millisecond)
	}
	proxy.resume()
	wait(t, client.Ready)
	history, _ := client.Events.Snapshot()
	resumed := false
	for _, e := range history {
		if e.Kind == events.ConnectionUp && strings.Contains(e.Outcome, "session_present=true") {
			resumed = true
		}
	}
	if resumed != expectResumed {
		t.Fatal("unexpected broker session present result")
	}
	ack, err := b.Command(ctx, backend.NewCommand("sim-001", "get_version", backend.ID(), 5*time.Second))
	if err != nil || ack == nil || ack.Result.Version == "" {
		t.Fatal("command did not recover after reconnect")
	}
	t.Log("actual connection loss -> unknown, TLS reconnect, session-dependent subscription restoration and command recovery: PASS")
}