feat: per-module build locking with a timeout (#1737)

backend/controller/sql/sqltest/testing.go

@@ -8,25 +8,21 @@ import (
 	"time"
 
 	"github.com/alecthomas/assert/v2"
-	"github.com/gofrs/flock"
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	"github.com/TBD54566975/ftl/backend/controller/sql/databasetesting"
+	"github.com/TBD54566975/ftl/internal/flock"
 )
 
 // OpenForTesting opens a database connection for testing, recreating the
 // database beforehand.
 func OpenForTesting(ctx context.Context, t testing.TB) *pgxpool.Pool {
 	t.Helper()
-	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
-	defer cancel()
 	// Acquire lock for this DB.
 	lockPath := filepath.Join(os.TempDir(), "ftl-db-test.lock")
-	lock := flock.New(lockPath)
-	ok, err := lock.TryLockContext(ctx, time.Second)
+	release, err := flock.Acquire(ctx, lockPath, 10*time.Second)
 	assert.NoError(t, err)
-	assert.True(t, ok, "could not acquire lock on %s", lockPath)
-	t.Cleanup(func() { _ = lock.Unlock() })
+	t.Cleanup(func() { _ = release() })
 
 	testDSN := "postgres://localhost:54320/ftl-test?user=postgres&password=secret&sslmode=disable"
 	conn, err := databasetesting.CreateForDevel(ctx, testDSN, true)

buildengine/build.go

@@ -6,19 +6,26 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 
 	"google.golang.org/protobuf/proto"
 
 	schemapb "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/schema"
 	"github.com/TBD54566975/ftl/backend/schema"
 	"github.com/TBD54566975/ftl/common/moduleconfig"
 	"github.com/TBD54566975/ftl/internal/errors"
+	"github.com/TBD54566975/ftl/internal/flock"
 	"github.com/TBD54566975/ftl/internal/log"
 	"github.com/TBD54566975/ftl/internal/slices"
 )
 
+const BuildLockTimeout = time.Minute
+
 // Build a project in the given directory given the schema and project config.
+//
 // For a module, this will build the module. For an external library, this will build stubs for imported modules.
+//
+// A lock file is used to ensure that only one build is running at a time.
 func Build(ctx context.Context, sch *schema.Schema, project Project, filesTransaction ModifyFilesTransaction) error {
 	switch project := project.(type) {
 	case Module:
@@ -31,6 +38,11 @@ func Build(ctx context.Context, sch *schema.Schema, project Project, filesTransa
 }
 
 func buildModule(ctx context.Context, sch *schema.Schema, module Module, filesTransaction ModifyFilesTransaction) error {
+	release, err := flock.Acquire(ctx, filepath.Join(module.Dir, ".ftl-build-lock"), BuildLockTimeout)
+	if err != nil {
+		return err
+	}
+	defer release() //nolint:errcheck
 	logger := log.FromContext(ctx).Scope(module.Module)
 	ctx = log.ContextWithLogger(ctx, logger)
 
@@ -40,7 +52,6 @@ func buildModule(ctx context.Context, sch *schema.Schema, module Module, filesTr
 	}
 
 	logger.Infof("Building module")
-	var err error
 	switch module.Language {
 	case "go":
 		err = buildGoModule(ctx, sch, module, filesTransaction)

go.mod

@@ -28,7 +28,6 @@ require (
 	github.com/docker/docker v26.1.3+incompatible
 	github.com/docker/go-connections v0.5.0
 	github.com/go-logr/logr v1.4.2
-	github.com/gofrs/flock v0.8.1
 	github.com/golang/protobuf v1.5.4
 	github.com/google/uuid v1.6.0
 	github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438
@@ -134,7 +133,7 @@ require (
 	github.com/swaggest/refl v1.3.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.27.0 // indirect
 	golang.org/x/crypto v0.24.0 // indirect
-	golang.org/x/sys v0.21.0 // indirect
+	golang.org/x/sys v0.21.0
 	golang.org/x/text v0.16.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240520151616-dc85e6b867a5 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect

internal/flock/flock.go

@@ -0,0 +1,66 @@
+package flock
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+var ErrLocked = errors.New("locked")
+
+// Acquire a lock on the given path.
+//
+// The lock is released when the returned function is called.
+func Acquire(ctx context.Context, path string, timeout time.Duration) (release func() error, err error) {
+	absPath, err := filepath.Abs(path)
+	if err != nil {
+		return nil, err
+	}
+	end := time.Now().Add(timeout)
+	for {
+		release, err := acquire(absPath)
+		if err == nil {
+			return release, nil
+		}
+		if !errors.Is(err, ErrLocked) {
+			return nil, fmt.Errorf("failed to acquire lock %s: %w", absPath, err)
+		}
+		if time.Now().After(end) {
+			pid, _ := os.ReadFile(absPath)
+			return nil, fmt.Errorf("timed out acquiring lock %s, locked by pid %s: %w", absPath, pid, err)
+		}
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-time.After(time.Second):
+		}
+	}
+}
+
+func acquire(path string) (release func() error, err error) {
+	pid := os.Getpid()
+	fd, err := unix.Open(path, unix.O_CREAT|unix.O_RDWR|unix.O_CLOEXEC|unix.O_SYNC, 0600)
+	if err != nil {
+		return nil, fmt.Errorf("open failed: %w", err)
+	}
+
+	err = unix.Flock(fd, unix.LOCK_EX|unix.LOCK_NB)
+	if err != nil {
+		_ = unix.Close(fd)
+		return nil, fmt.Errorf("%w: %w", ErrLocked, err)
+	}
+
+	_, err = unix.Write(fd, []byte(strconv.Itoa(pid)))
+	if err != nil {
+		return nil, fmt.Errorf("write failed: %w", err)
+	}
+	return func() error {
+		return errors.Join(unix.Flock(fd, unix.LOCK_UN), unix.Close(fd), os.Remove(path))
+	}, nil
+}

internal/flock/flock_test.go

@@ -0,0 +1,28 @@
+package flock
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+
+	"github.com/alecthomas/assert/v2"
+)
+
+func TestFlock(t *testing.T) {
+	dir := t.TempDir()
+	lockfile := filepath.Join(dir, "lock")
+	ctx := context.Background()
+	release, err := Acquire(ctx, lockfile, 0)
+	assert.NoError(t, err)
+
+	_, err = Acquire(ctx, lockfile, 0)
+	assert.Error(t, err)
+
+	err = release()
+	assert.NoError(t, err)
+
+	releaseb, err := Acquire(ctx, lockfile, 0)
+	assert.NoError(t, err)
+	err = releaseb()
+	assert.NoError(t, err)
+}