fix: prevent db error if no async calls to execute (#2081)

When a controller tries to get a lease on the next async call to
execute, the lease creation code should first check to see if there was
a call found.
Before when a controller tried to get a hold on the next async call to
execute, it would result in this error:
```
2024-07-15 16:00:08 UTC:10.2.111.21(46692):adminuser@tbd:[60308]:ERROR:  null value in column "key" of relation "leases" violates not-null constraint
2024-07-15 16:00:08 UTC:10.2.111.21(46692):adminuser@tbd:[60308]:DETAIL:  Failing row contains (1632753, 37a8d114-73ca-46a1-b03b-a17331d17181, null, 2024-07-15 16:00:08.260772+00, 2024-07-15 16:00:13.260772+00, null).
2024-07-15 16:00:08 UTC:10.2.111.21(46692):adminuser@tbd:[60308]:STATEMENT:  -- name: AcquireAsyncCall :one
	WITH async_call AS (
	  SELECT id
	  FROM async_calls
	  WHERE state = 'pending' AND scheduled_at <= (NOW() AT TIME ZONE 'utc')
	  LIMIT 1
	  FOR UPDATE SKIP LOCKED
	), lease AS (
	  INSERT INTO leases (idempotency_key, key, expires_at)
	  VALUES (gen_random_uuid(), '/system/async_call/' || (SELECT id FROM async_call), (NOW() AT TIME ZONE 'utc') + $1::interval)
	  RETURNING id, idempotency_key, key, created_at, expires_at, metadata
	)
	UPDATE async_calls
	SET state = 'executing', lease_id = (SELECT id FROM lease)
	WHERE id = (SELECT id FROM async_call)
	RETURNING
	  id AS async_call_id,
	  (SELECT idempotency_key FROM lease) AS lease_idempotency_key,
	  (SELECT key FROM lease) AS lease_key,
	  origin,
	  verb,
	  request,
	  scheduled_at,
	  remaining_attempts,
	  backoff,
	  max_backoff
```

backend/controller/dal/async_calls.go

@@ -96,8 +96,7 @@ func (d *DAL) AcquireAsyncCall(ctx context.Context) (call *AsyncCall, err error)
 	row, err := tx.db.AcquireAsyncCall(ctx, ttl)
 	if err != nil {
 		err = dalerrs.TranslatePGError(err)
-		// We get a NULL constraint violation if there are no async calls to acquire, so translate it to ErrNotFound.
-		if errors.Is(err, dalerrs.ErrConstraint) {
+		if errors.Is(err, dalerrs.ErrNotFound) {
 			return nil, fmt.Errorf("no pending async calls: %w", dalerrs.ErrNotFound)
 		}
 		return nil, fmt.Errorf("failed to acquire async call: %w", err)

backend/controller/dal/async_calls_test.go

@@ -0,0 +1,22 @@
+package dal
+
+import (
+	"context"
+	"testing"
+
+	"github.com/TBD54566975/ftl/backend/controller/sql/sqltest"
+	dalerrs "github.com/TBD54566975/ftl/backend/dal"
+	"github.com/TBD54566975/ftl/internal/log"
+	"github.com/alecthomas/assert/v2"
+)
+
+func TestNoCallToAcquire(t *testing.T) {
+	ctx := log.ContextWithNewDefaultLogger(context.Background())
+	conn := sqltest.OpenForTesting(ctx, t)
+	dal, err := New(ctx, conn)
+	assert.NoError(t, err)
+
+	_, err = dal.AcquireAsyncCall(ctx)
+	assert.IsError(t, err, dalerrs.ErrNotFound)
+	assert.EqualError(t, err, "no pending async calls: not found")
+}

backend/controller/sql/queries.sql

@@ -456,7 +456,8 @@ WITH async_call AS (
   FOR UPDATE SKIP LOCKED
 ), lease AS (
   INSERT INTO leases (idempotency_key, key, expires_at)
-  VALUES (gen_random_uuid(), '/system/async_call/' || (SELECT id FROM async_call), (NOW() AT TIME ZONE 'utc') + @ttl::interval)
+  SELECT gen_random_uuid(), '/system/async_call/' || (SELECT id FROM async_call), (NOW() AT TIME ZONE 'utc') + @ttl::interval
+  WHERE (SELECT id FROM async_call) IS NOT NULL
   RETURNING *
 )
 UPDATE async_calls