chore(ci): multiple integration tests per shard in CI

.github/workflows/ci.yml

@@ -257,7 +257,7 @@ jobs:
         run: |
           set -euo pipefail
           # shellcheck disable=SC2046
-          echo "matrix={\"test\":$(jq -c -n '$ARGS.positional' --args $(git grep -l '^//go:build integration' | xargs grep '^func Test' | awk '{print $2}' | cut -d'(' -f1))}" >> "$GITHUB_OUTPUT"
+          echo "matrix={\"test\":$(jq -c -n '$ARGS.positional' --args $(git grep -l '^//go:build integration' | xargs grep '^func Test' | awk '{print $2}' | cut -d'(' -f1 | paste -d ' ' - - - | sed 's/|*$//'))}" >> "$GITHUB_OUTPUT"
   integration-run:
     name: Integration Test
     # if: github.event_name != 'pull_request' || github.event.action == 'enqueued' || contains( github.event.pull_request.labels.*.name, 'run-all')
@@ -281,7 +281,7 @@ jobs:
         run: |
           set -euo pipefail
           # shellcheck disable=SC2046
-          go test -v -race -tags integration -run '^${{ matrix.test }}$' $(git grep -l '^//go:build integration' | xargs grep -l '^func ${{ matrix.test }}' | xargs -I {} dirname ./{})
+          echo ${{ matrix.test }} | xargs -n1 just integration-tests
   infrastructure-shard:
     name: Shard Infrastructure Tests
     # if: github.event_name != 'pull_request' || github.event.action == 'enqueued' || contains( github.event.pull_request.labels.*.name, 'run-all')

internal/integration/actions.go

@@ -7,7 +7,6 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
-	"fmt"
 	"io"
 	"net/http"
 	"net/url"
@@ -624,7 +623,7 @@ func JsonData(t testing.TB, body interface{}) []byte {
 func HttpCall(method string, path string, headers map[string][]string, body []byte, onResponse func(t testing.TB, resp *HTTPResponse)) Action {
 	return func(t testing.TB, ic TestContext) {
 		Infof("HTTP %s %s", method, path)
-		baseURL, err := url.Parse(fmt.Sprintf("http://localhost:8891"))
+		baseURL, err := url.Parse("http://localhost:8891")
 		assert.NoError(t, err)
 
 		u, err := baseURL.Parse(path)

internal/integration/harness.go

@@ -38,7 +38,6 @@ import (
 	"github.com/block/ftl/backend/provisioner/scaling/k8sscaling"
 	"github.com/block/ftl/internal"
 	"github.com/block/ftl/internal/exec"
-	ftlexec "github.com/block/ftl/internal/exec"
 	"github.com/block/ftl/internal/log"
 	"github.com/block/ftl/internal/rpc"
 )
@@ -64,9 +63,11 @@ func Infof(format string, args ...any) {
 	fmt.Printf("\033[32m\033[1mINFO: "+format+"\033[0m\n", args...)
 }
 
-var buildOnce sync.Once
-
-var buildOnceOptions *options
+var buildOnceKube sync.Once
+var buildOnceConsole sync.Once
+var buildOnceWithoutConsole sync.Once
+var buildJVMOnce sync.Once
+var startLocalStackOnce sync.Once
 
 // An Option for configuring the integration test harness.
 type Option func(*options)
@@ -256,62 +257,67 @@ func run(t *testing.T, actionsOrOptions ...ActionOrOption) {
 
 	var kubeClient *kubernetes.Clientset
 	var kubeNamespace string
-	buildOnce.Do(func() {
-		buildOnceOptions = &opts
-		if opts.kube {
+	if opts.kube {
+		buildOnceKube.Do(func() {
 			// This command will build a linux/amd64 version of FTL and deploy it to the kube cluster
 			Infof("Building FTL and deploying to kube")
-			err = ftlexec.Command(ctx, log.Debug, filepath.Join(rootDir, "deployment"), "just", "setup-istio-cluster").RunBuffered(ctx)
+			err = exec.Command(ctx, log.Debug, filepath.Join(rootDir, "deployment"), "just", "setup-istio-cluster").RunBuffered(ctx)
 			assert.NoError(t, err)
 
 			// On CI we always skip the full deploy, as the build is done in the CI pipeline
 			skipKubeFullDeploy := os.Getenv("CI") != ""
 			if skipKubeFullDeploy {
 				Infof("Skipping full deploy since CI is set")
 			} else {
-				err = ftlexec.Command(ctx, log.Debug, filepath.Join(rootDir, "deployment"), "just", "full-deploy").RunBuffered(ctx)
+				err = exec.Command(ctx, log.Debug, filepath.Join(rootDir, "deployment"), "just", "full-deploy").RunBuffered(ctx)
 				assert.NoError(t, err)
 			}
 			if runtime.GOOS != "linux" || runtime.GOARCH != "amd64" {
 				// If we are already on linux/amd64 we don't need to rebuild, otherwise we now need a native one to interact with the kube cluster
 				Infof("Building FTL for native OS")
-				err = ftlexec.Command(ctx, log.Debug, rootDir, "just", "build", "ftl").RunBuffered(ctx)
+				err = exec.Command(ctx, log.Debug, rootDir, "just", "build", "ftl").RunBuffered(ctx)
 				assert.NoError(t, err)
 			}
 			kubeClient, err = k8sscaling.CreateClientSet()
 			assert.NoError(t, err)
 			kubeNamespace, err = k8sscaling.GetCurrentNamespace()
 			assert.NoError(t, err)
 			// We create the client before, as kube itself is up, we are just waiting for the deployments
-			err = ftlexec.Command(ctx, log.Debug, filepath.Join(rootDir, "deployment"), "just", "wait-for-kube").RunBuffered(ctx)
+			err = exec.Command(ctx, log.Debug, filepath.Join(rootDir, "deployment"), "just", "wait-for-kube").RunBuffered(ctx)
 			if err != nil {
 				dumpKubePods(ctx, optional.Ptr(kubeClient), kubeNamespace)
 			}
 			assert.NoError(t, err)
 			ver := os.Getenv("OLD_FTL_VERSION")
 			if ver != "" {
-				err = ftlexec.Command(ctx, log.Debug, filepath.Join(rootDir, "deployment"), "just", "wait-for-version-upgrade", ver).RunBuffered(ctx)
+				err = exec.Command(ctx, log.Debug, filepath.Join(rootDir, "deployment"), "just", "wait-for-version-upgrade", ver).RunBuffered(ctx)
 			}
-		} else if opts.console {
+		})
+	} else if opts.console {
+		buildOnceConsole.Do(func() {
 			Infof("Building ftl with console")
-			err = ftlexec.Command(ctx, log.Debug, rootDir, "just", "build", "ftl").RunBuffered(ctx)
+			err = exec.Command(ctx, log.Debug, rootDir, "just", "build", "ftl").RunBuffered(ctx)
 			assert.NoError(t, err)
-		} else {
+		})
+	} else {
+		buildOnceWithoutConsole.Do(func() {
 			Infof("Building ftl without console")
-			err = ftlexec.Command(ctx, log.Debug, rootDir, "just", "build-without-frontend", "ftl").RunBuffered(ctx)
+			err = exec.Command(ctx, log.Debug, rootDir, "just", "build-without-frontend", "ftl").RunBuffered(ctx)
 			assert.NoError(t, err)
-		}
-		if opts.requireJava || slices.Contains(opts.languages, "java") || slices.Contains(opts.languages, "kotlin") {
-			err = ftlexec.Command(ctx, log.Debug, rootDir, "just", "build-jvm", "-DskipTests", "-B").RunBuffered(ctx)
+		})
+	}
+	if opts.requireJava || slices.Contains(opts.languages, "java") || slices.Contains(opts.languages, "kotlin") {
+		buildJVMOnce.Do(func() {
+			err = exec.Command(ctx, log.Debug, rootDir, "just", "build-jvm", "-DskipTests", "-B").RunBuffered(ctx)
 			assert.NoError(t, err)
-		}
-		if opts.localstack {
-			err = ftlexec.Command(ctx, log.Debug, rootDir, "just", "localstack").RunBuffered(ctx)
+		})
+	}
+	if opts.localstack {
+		startLocalStackOnce.Do(func() {
+			err = exec.Command(ctx, log.Debug, rootDir, "just", "localstack").RunBuffered(ctx)
 			assert.NoError(t, err)
-		}
-	})
-
-	assert.Equal(t, *buildOnceOptions, opts, "Options changed between test runs")
+		})
+	}
 
 	for _, language := range opts.languages {
 		ctx, done := context.WithCancel(ctx)
@@ -618,7 +624,7 @@ func (l *logWriter) Write(p []byte) (n int, err error) {
 func startProcess(ctx context.Context, t testing.TB, tempDir string, devMode bool, args ...string) context.Context {
 	t.Helper()
 	ctx, cancel := context.WithCancel(ctx)
-	cmd := ftlexec.Command(ctx, log.Debug, "..", args[0], args[1:]...)
+	cmd := exec.Command(ctx, log.Debug, "..", args[0], args[1:]...)
 	if devMode {
 		cmd.Dir = tempDir
 	}