feat(mysql): Wire MySQL to controller and add integration test

Author: sbalabanov-zz

MODULE.bazel

@@ -33,6 +33,8 @@ use_repo(
     "com_github_go_sql_driver_mysql",
     "com_github_gogo_protobuf",
     "com_github_stretchr_testify",
+    "com_github_testcontainers_testcontainers_go",
+    "com_github_testcontainers_testcontainers_go_modules_mysql",
     "com_github_uber_go_tally_v4",
     "org_golang_google_grpc",
     "org_golang_google_protobuf",

Makefile

@@ -68,10 +68,10 @@ integration-test:
 	@echo "Running all service integration tests..."
 	@$(BAZEL) test //gateway/integration_tests:integration_tests_test //orchestrator/integration_tests:integration_tests_test //speculator/integration_tests:integration_tests_test --test_output=all
 
-# Run end-to-end tests (requires all services to be running)
+# Run end-to-end integration tests (hermetic, no manual server setup needed)
 e2e-test:
-	@echo "Running end-to-end tests..."
-	@$(BAZEL) test //integration_tests:e2e_test --test_output=all
+	@echo "Running integration tests..."
+	@$(BAZEL) test //integration_tests:integration_test --test_output=all
 
 # Clean generated files and binaries
 clean:
@@ -186,7 +186,9 @@ help:
 	@echo "  make integration-test-orchestrator - Test Orchestrator service"
 	@echo "  make integration-test-speculator   - Test Speculator service"
 	@echo "  make integration-test   - Test all services"
-	@echo "  make e2e-test           - Run end-to-end tests"
+	@echo ""
+	@echo "End-to-End Tests (hermetic, no setup needed):"
+	@echo "  make e2e-test           - Run integration tests with Testcontainers"
 	@echo ""
 	@echo "Run Clients:"
 	@echo "  make run-client-gateway - Run gateway client"

examples/server/gateway/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "github.com/uber/submitqueue/examples/server/gateway",
     visibility = ["//visibility:private"],
     deps = [
+        "//extensions/storage/mysql",
         "//gateway/controller",
         "//gateway/protopb",
         "@com_github_uber_go_tally_v4//:tally",

examples/server/gateway/main.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/uber-go/tally/v4"
+	"github.com/uber/submitqueue/extensions/storage/mysql"
 	"github.com/uber/submitqueue/gateway/controller"
 	pb "github.com/uber/submitqueue/gateway/protopb"
 	"go.uber.org/zap"
@@ -21,12 +22,18 @@ import (
 // GatewayServer wraps the controller and implements the gRPC service interface
 type GatewayServer struct {
 	pb.UnimplementedSubmitQueueGatewayServer
-	controller *controller.PingController
+	pingController *controller.PingController
+	landController *controller.LandController
 }
 
 // Ping delegates to the controller
 func (s *GatewayServer) Ping(ctx context.Context, req *pb.PingRequest) (*pb.PingResponse, error) {
-	return s.controller.Ping(ctx, req)
+	return s.pingController.Ping(ctx, req)
+}
+
+// Land delegates to the controller
+func (s *GatewayServer) Land(ctx context.Context, req *pb.LandRequest) (*pb.LandResponse, error) {
+	return s.landController.Land(ctx, req)
 }
 
 func main() {
@@ -75,21 +82,42 @@ func run() error {
 		metricsWgDone.Wait()
 	}()
 
+	// Initialize MySQL storage factory
+	mysqlDSN := os.Getenv("MYSQL_DSN")
+	if mysqlDSN == "" {
+		mysqlDSN = "root:root@tcp(localhost:3306)/submitqueue?parseTime=true"
+	}
+	storeFactory, err := mysql.NewFactory(mysql.MySQLParameters{
+		DSN:             mysqlDSN,
+		MaxOpenConns:    10,
+		MaxIdleConns:    5,
+		ConnMaxLifetime: 5 * time.Minute,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create MySQL storage factory: %w", err)
+	}
+	defer storeFactory.Close()
+
 	// Create gRPC server
 	grpcServer := grpc.NewServer()
 
-	// Create ping controller and wrap it for gRPC
+	// Create controllers and wrap them for gRPC
 	pingController := controller.NewPingController(logger, scope)
+	landController := controller.NewLandController(logger, scope, storeFactory)
 	gatewayServer := &GatewayServer{
-		controller: pingController,
+		pingController: pingController,
+		landController: landController,
 	}
 	pb.RegisterSubmitQueueGatewayServer(grpcServer, gatewayServer)
 
 	// Register reflection service for debugging with grpcurl
 	reflection.Register(grpcServer)
 
-	// Listen on port 8081
-	port := ":8081"
+	// Listen on configurable port
+	port := os.Getenv("PORT")
+	if port == "" {
+		port = ":8081"
+	}
 	listener, err := net.Listen("tcp", port)
 	if err != nil {
 		return fmt.Errorf("failed to listen on port %s: %w", port, err)

examples/server/orchestrator/main.go

@@ -88,8 +88,11 @@ func run() error {
 	// Register reflection service for debugging with grpcurl
 	reflection.Register(grpcServer)
 
-	// Listen on port 8082
-	port := ":8082"
+	// Listen on configurable port
+	port := os.Getenv("PORT")
+	if port == "" {
+		port = ":8082"
+	}
 	listener, err := net.Listen("tcp", port)
 	if err != nil {
 		return fmt.Errorf("failed to listen on port %s: %w", port, err)

examples/server/speculator/main.go

@@ -88,8 +88,11 @@ func run() error {
 	// Register reflection service for debugging with grpcurl
 	reflection.Register(grpcServer)
 
-	// Listen on port 8083
-	port := ":8083"
+	// Listen on configurable port
+	port := os.Getenv("PORT")
+	if port == "" {
+		port = ":8083"
+	}
 	listener, err := net.Listen("tcp", port)
 	if err != nil {
 		return fmt.Errorf("failed to listen on port %s: %w", port, err)

extensions/storage/mysql/schema/BUILD.bazel

@@ -0,0 +1,5 @@
+filegroup(
+    name = "schema",
+    srcs = glob(["*.sql"]),
+    visibility = ["//visibility:public"],
+)

extensions/storage/mysql/schema/request.sql

@@ -3,8 +3,8 @@ CREATE TABLE IF NOT EXISTS request (
     seq BIGINT NOT NULL,
     change_source VARCHAR(255) NOT NULL,
     change_ids JSON NOT NULL,
-    land_strategy INT NOT NULL DEFAULT 0,
+    land_strategy INT NOT NULL,
     state INT NOT NULL,
-    version INT NOT NULL DEFAULT 1,
+    version INT NOT NULL,
     PRIMARY KEY (queue, seq)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

gateway/controller/BUILD.bazel

@@ -9,6 +9,8 @@ go_library(
     importpath = "github.com/uber/submitqueue/gateway/controller",
     visibility = ["//visibility:public"],
     deps = [
+        "//entities",
+        "//extensions/storage",
         "//gateway/protopb",
         "@com_github_uber_go_tally_v4//:tally",
         "@org_uber_go_zap//:zap",
@@ -23,6 +25,8 @@ go_test(
     ],
     embed = [":controller"],
     deps = [
+        "//entities",
+        "//extensions/storage",
         "//gateway/protopb",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",

gateway/controller/land.go

@@ -6,6 +6,8 @@ import (
 	"time"
 
 	"github.com/uber-go/tally/v4"
+	"github.com/uber/submitqueue/entities"
+	"github.com/uber/submitqueue/extensions/storage"
 	pb "github.com/uber/submitqueue/gateway/protopb"
 	"go.uber.org/zap"
 )
@@ -14,13 +16,15 @@ import (
 type LandController struct {
 	logger       *zap.Logger
 	metricsScope tally.Scope
+	storeFactory storage.StoreFactory
 }
 
 // NewLandController creates a new instance of the gateway land controller
-func NewLandController(logger *zap.Logger, scope tally.Scope) *LandController {
+func NewLandController(logger *zap.Logger, scope tally.Scope, storeFactory storage.StoreFactory) *LandController {
 	return &LandController{
 		logger:       logger,
 		metricsScope: scope,
+		storeFactory: storeFactory,
 	}
 }
 
@@ -33,8 +37,18 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (*pb.Lan
 
 	c.metricsScope.Counter("land_request_count").Inc(1)
 
-	// TODO: Implement proper SQID generation and send the request to the appropriate queue. So far unix time to make it sequential.
-	sqid := fmt.Sprintf("%d", time.Now().Unix())
+	change := entities.Change{
+		Source: req.Change.GetSource(),
+		IDs:    req.Change.GetIds(),
+	}
+	strategy := entities.RequestLandStrategy(int(req.Strategy))
+
+	request, err := c.storeFactory.GetRequestStore().Create(ctx, req.Queue, change, strategy, entities.RequestStateNew)
+	if err != nil {
+		return nil, fmt.Errorf("LandController failed to create request for queue=%s: %w", req.Queue, err)
+	}
+
+	sqid := request.GetID()
 
 	c.logger.Debug("land request received",
 		zap.String("queue", req.Queue),