fix: stabilize federated agent runtime updates (#550)

go-gost/x/socket/command_dispatch_test.go

@@ -0,0 +1,117 @@
+package socket
+
+import (
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	coreservice "github.com/go-gost/core/service"
+	"github.com/go-gost/x/config"
+	"github.com/go-gost/x/registry"
+)
+
+type blockingCommandService struct {
+	started     chan struct{}
+	release     chan struct{}
+	startedOnce sync.Once
+}
+
+func (s *blockingCommandService) Serve() error   { return nil }
+func (s *blockingCommandService) Addr() net.Addr { return nil }
+func (s *blockingCommandService) Close() error {
+	s.startedOnce.Do(func() { close(s.started) })
+	<-s.release
+	return nil
+}
+
+func TestMutationCommandsAreSerialized(t *testing.T) {
+	mutations := []string{
+		"AddService", "UpdateService", "DeleteService", "PauseService", "ResumeService",
+		"AddChains", "UpdateChains", "DeleteChains",
+		"AddLimiters", "UpdateLimiters", "DeleteLimiters",
+		"AddCLimiters", "UpdateCLimiters", "DeleteCLimiters",
+		"SetProtocol", "UpgradeAgent", "RollbackAgent", "reload",
+	}
+	for _, command := range mutations {
+		if !isMutationCommand(command) {
+			t.Fatalf("expected %s to use the serialized mutation queue", command)
+		}
+	}
+}
+
+func TestReadOnlyCommandsRemainBoundedAsync(t *testing.T) {
+	for _, command := range []string{"TcpPing", "UdpPing", "ServiceMonitorCheck"} {
+		if isMutationCommand(command) {
+			t.Fatalf("expected %s to remain a read-only command", command)
+		}
+	}
+}
+
+func TestCommandResponseTypePreservesRequestContract(t *testing.T) {
+	if got := commandResponseType("UpdateService"); got != "UpdateServiceResponse" {
+		t.Fatalf("unexpected response type: %s", got)
+	}
+	if got := commandResponseType(""); got != "UnknownCommandResponse" {
+		t.Fatalf("unexpected empty command response type: %s", got)
+	}
+}
+
+func TestMutationQueueExecutesCommandsInArrivalOrder(t *testing.T) {
+	originalConfig := config.Global()
+	t.Cleanup(func() { config.Set(originalConfig) })
+
+	firstName := "mutation_queue_first_tdd"
+	secondName := "mutation_queue_second_tdd"
+	first := &blockingCommandService{started: make(chan struct{}), release: make(chan struct{})}
+	second := &blockingCommandService{started: make(chan struct{}), release: make(chan struct{})}
+	for _, name := range []string{firstName, secondName} {
+		registry.ServiceRegistry().Unregister(name)
+	}
+	t.Cleanup(func() {
+		select {
+		case <-first.release:
+		default:
+			close(first.release)
+		}
+		select {
+		case <-second.release:
+		default:
+			close(second.release)
+		}
+		registry.ServiceRegistry().Unregister(firstName)
+		registry.ServiceRegistry().Unregister(secondName)
+	})
+	if err := registry.ServiceRegistry().Register(firstName, coreservice.Service(first)); err != nil {
+		t.Fatalf("register first service: %v", err)
+	}
+	if err := registry.ServiceRegistry().Register(secondName, coreservice.Service(second)); err != nil {
+		t.Fatalf("register second service: %v", err)
+	}
+	config.Set(&config.Config{Services: []*config.ServiceConfig{{Name: firstName}, {Name: secondName}}})
+
+	reporter := NewWebSocketReporter("", "mutation-queue-test-secret")
+	go reporter.runMutationCommands()
+	t.Cleanup(reporter.Stop)
+	reporter.dispatchCommand(CommandMessage{Type: "DeleteService", Data: map[string]any{"services": []string{firstName}}})
+	reporter.dispatchCommand(CommandMessage{Type: "DeleteService", Data: map[string]any{"services": []string{secondName}}})
+
+	select {
+	case <-first.started:
+	case <-time.After(time.Second):
+		t.Fatal("first mutation did not start")
+	}
+	select {
+	case <-second.started:
+		t.Fatal("second mutation started before first mutation completed")
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	close(first.release)
+	select {
+	case <-second.started:
+	case <-time.After(time.Second):
+		t.Fatal("second mutation did not start after first mutation completed")
+	}
+	close(second.release)
+}