Merge branch 'main' into jonathanj/otel/fsm2

.github/workflows/autofmt.yml → .github/workflows/autofmtpr.yml

@@ -1,16 +1,13 @@
 on:
-  schedule:
-    # Run daily
-    - cron: "0 22 * * *"
-name: Auto-format
+  pull_request:
+name: Auto-format Pull Request
 jobs:
   format:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
         with:
-          ref: main
-          fetch-depth: 0
+          ref: ${{ github.head_ref }}
           token: ${{ secrets.FTL_AUTOFMT_TOKEN }}
       - name: Init Hermit
         run: ./bin/hermit env -r >> "$GITHUB_ENV"
@@ -20,9 +17,11 @@ jobs:
         id: git-check
         run: echo "modified=$(if git diff-index --quiet HEAD --; then echo "false"; else echo "true"; fi)" >> "$GITHUB_OUTPUT"
       - name: Push changes
+        env:
+          HEAD: ${{ github.head_ref }}
         if: steps.git-check.outputs.modified == 'true'
         run: |
           git config --global user.email "github-actions[bot]@users.noreply.github.com"
           git config --global user.name "github-actions[bot]"
           git commit -am "chore(autofmt): Automated formatting"
-          git push
+          git push origin "$HEAD"

.golangci.yml

@@ -93,6 +93,8 @@ linters-settings:
             desc: "use github.com/TBD54566975/ftl/internal/exec"
           - pkg: golang.design/x/reflect
             desc: "use github.com/TBD54566975/ftl/internal/reflect"
+          - pkg: "github.com/reugn/go-quartz/logger"
+            desc: "use github.com/TBD54566975/ftl/internal/log"
   # wrapcheck:
   #   ignorePackageGlobs:
   #     - github.com/TBD54566975/ftl/*

backend/controller/admin/client.go

@@ -8,6 +8,7 @@ import (
 	"net/url"
 
 	"connectrpc.com/connect"
+
 	ftlv1 "github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1"
 	"github.com/TBD54566975/ftl/backend/protos/xyz/block/ftl/v1/ftlv1connect"
 )

backend/controller/admin/local_client.go

@@ -5,11 +5,12 @@ import (
 	"fmt"
 	"path/filepath"
 
+	"github.com/alecthomas/types/optional"
+
 	"github.com/TBD54566975/ftl/backend/schema"
 	"github.com/TBD54566975/ftl/buildengine"
 	cf "github.com/TBD54566975/ftl/common/configuration"
 	"github.com/TBD54566975/ftl/common/projectconfig"
-	"github.com/alecthomas/types/optional"
 )
 
 // localClient reads and writes to local projectconfig files without making any network

backend/controller/console.go

@@ -7,8 +7,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/reugn/go-quartz/logger"
-
 	"connectrpc.com/connect"
 	"github.com/alecthomas/types/optional"
 	"google.golang.org/protobuf/types/known/durationpb"
@@ -165,7 +163,7 @@ func (c *ConsoleService) GetModules(ctx context.Context, req *connect.Request[pb
 
 	sorted, err := buildengine.TopologicalSort(graph(sch))
 	if err != nil {
-		logger.Debugf(err.Error())
+		return nil, fmt.Errorf("failed to sort modules: %w", err)
 	}
 	topology := &pbconsole.Topology{
 		Levels: make([]*pbconsole.TopologyGroup, len(sorted)),

backend/controller/controller.go

@@ -7,7 +7,6 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
-	"github.com/TBD54566975/ftl/backend/controller/observability"
 	"hash"
 	"io"
 	"math/rand"
@@ -43,6 +42,7 @@ import (
 	"github.com/TBD54566975/ftl/backend/controller/dal"
 	"github.com/TBD54566975/ftl/backend/controller/ingress"
 	"github.com/TBD54566975/ftl/backend/controller/leases"
+	"github.com/TBD54566975/ftl/backend/controller/observability"
 	"github.com/TBD54566975/ftl/backend/controller/pubsub"
 	"github.com/TBD54566975/ftl/backend/controller/scaling"
 	"github.com/TBD54566975/ftl/backend/controller/scaling/localscaling"

backend/controller/cronjobs/cronjobs.go

@@ -307,7 +307,7 @@ func (s *Service) watchForUpdates(ctx context.Context) {
 		now := s.clock.Now()
 		next := now.Add(time.Hour) // should never be reached, expect a different signal long beforehand
 		for _, j := range state.jobs {
-			if possibleNext, err := s.nextAttemptForJob(j, state, false); err == nil && possibleNext.Before(next) {
+			if possibleNext, err := s.nextAttemptForJob(j, state, now, false); err == nil && possibleNext.Before(next) {
 				next = possibleNext
 			}
 		}
@@ -329,8 +329,11 @@ func (s *Service) watchForUpdates(ctx context.Context) {
 			return
 		case <-s.clock.After(next.Sub(now)):
 			// Try starting jobs in db
+			// note that we use next here are the current time
+			// as if there is a pause of over a second we could miss jobs if we use the current time
+			// this is very unlikely to happen, but if it did it would be hard to diagnose
 			jobsToAttempt := slices.Filter(state.jobs, func(j model.CronJob) bool {
-				if n, err := s.nextAttemptForJob(j, state, true); err == nil {
+				if n, err := s.nextAttemptForJob(j, state, next, true); err == nil {
 					return !n.After(s.clock.Now().UTC())
 				}
 				return false
@@ -386,7 +389,8 @@ func (s *Service) watchForUpdates(ctx context.Context) {
 	}
 }
 
-func (s *Service) nextAttemptForJob(job model.CronJob, state *state, allowsNow bool) (time.Time, error) {
+func (s *Service) nextAttemptForJob(job model.CronJob, state *state, currentTime time.Time, allowsNow bool) (time.Time, error) {
+	currentTime = currentTime.UTC()
 	if !s.isResponsibleForJob(job, state) {
 		return s.clock.Now(), fmt.Errorf("controller is not responsible for job")
 	}
@@ -401,7 +405,7 @@ func (s *Service) nextAttemptForJob(job model.CronJob, state *state, allowsNow b
 		if err != nil {
 			return s.clock.Now(), fmt.Errorf("failed to parse cron schedule %q", job.Schedule)
 		}
-		next, err := cron.NextAfter(pattern, s.clock.Now().UTC(), allowsNow)
+		next, err := cron.NextAfter(pattern, currentTime, allowsNow)
 		if err == nil {
 			return next, nil
 		}

backend/controller/cronjobs/testdata/go/cron/go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
+	github.com/hashicorp/cronexpr v1.1.2 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
 	github.com/jackc/pgx/v5 v5.6.0 // indirect

backend/controller/dal/fsm.go

@@ -5,12 +5,12 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/TBD54566975/ftl/backend/controller/observability"
 	"time"
 
 	"github.com/alecthomas/types/optional"
 
 	"github.com/TBD54566975/ftl/backend/controller/leases"
+	"github.com/TBD54566975/ftl/backend/controller/observability"
 	"github.com/TBD54566975/ftl/backend/controller/sql"
 	dalerrs "github.com/TBD54566975/ftl/backend/dal"
 	"github.com/TBD54566975/ftl/backend/schema"

backend/controller/dal/testdata/go/fsm/go.mod

@@ -26,12 +26,12 @@ require (
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.13 // indirect
 	github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.32.1 // indirect
 	github.com/aws/smithy-go v1.20.3 // indirect
-	github.com/benbjohnson/clock v1.3.5 // indirect
 	github.com/danieljoos/wincred v1.2.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/hashicorp/cronexpr v1.1.2 // indirect
 	github.com/hexops/gotextdiff v1.0.3 // indirect
 	github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect

backend/controller/dal/testdata/go/fsmretry/go.mod

@@ -20,6 +20,7 @@ require (
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
+	github.com/hashicorp/cronexpr v1.1.2 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
 	github.com/jackc/pgx/v5 v5.6.0 // indirect

backend/controller/ingress/handler.go

@@ -35,12 +35,15 @@ func Handle(
 			http.NotFound(w, r)
 			return
 		}
-		http.Error(w, err.Error(), http.StatusInternalServerError)
+		logger.Errorf(err, "failed to resolve route for %s %s", r.Method, r.URL.Path)
+		http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
 		return
 	}
 
 	body, err := BuildRequestBody(route, r, sch)
 	if err != nil {
+		// Only log at debug, as this is a client side error
+		logger.Debugf("bad request: %s", err.Error())
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
@@ -53,10 +56,12 @@ func Handle(
 
 	resp, err := call(r.Context(), creq, optional.Some(requestKey), r.RemoteAddr)
 	if err != nil {
+		logger.Errorf(err, "failed to call verb %s", route.Verb)
 		if connectErr := new(connect.Error); errors.As(err, &connectErr) {
-			http.Error(w, err.Error(), connectCodeToHTTP(connectErr.Code()))
+			httpCode := connectCodeToHTTP(connectErr.Code())
+			http.Error(w, http.StatusText(httpCode), httpCode)
 		} else {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
+			http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
 		}
 		return
 	}
@@ -65,22 +70,25 @@ func Handle(
 		verb := &schema.Verb{}
 		err = sch.ResolveToType(&schema.Ref{Name: route.Verb, Module: route.Module}, verb)
 		if err != nil {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
+			logger.Errorf(err, "could not resolve schema type for verb %s", route.Verb)
+			http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
 			return
 		}
 		var responseBody []byte
 
 		if metadata, ok := verb.GetMetadataIngress().Get(); ok && metadata.Type == "http" {
 			var response HTTPResponse
 			if err := json.Unmarshal(msg.Body, &response); err != nil {
-				http.Error(w, err.Error(), http.StatusInternalServerError)
+				logger.Errorf(err, "could not unmarhal response for verb %s", verb)
+				http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
 				return
 			}
 
 			var responseHeaders http.Header
 			responseBody, responseHeaders, err = ResponseForVerb(sch, verb, response)
 			if err != nil {
-				http.Error(w, err.Error(), http.StatusInternalServerError)
+				logger.Errorf(err, "could not create response for verb %s", verb)
+				http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
 				return
 			}
 
@@ -102,7 +110,7 @@ func Handle(
 		}
 
 	case *ftlv1.CallResponse_Error_:
-		http.Error(w, msg.Error.Message, http.StatusInternalServerError)
+		http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
 	}
 }
 

backend/controller/ingress/request.go

@@ -9,9 +9,10 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/alecthomas/types/optional"
+
 	"github.com/TBD54566975/ftl/backend/controller/dal"
 	"github.com/TBD54566975/ftl/backend/schema"
-	"github.com/alecthomas/types/optional"
 )
 
 // BuildRequestBody extracts the HttpRequest body from an HTTP request.

backend/controller/ingress/testdata/go/httpingress/go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
+	github.com/hashicorp/cronexpr v1.1.2 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
 	github.com/jackc/pgx/v5 v5.6.0 // indirect

backend/controller/leases/testdata/go/leases/go.mod

@@ -27,12 +27,12 @@ require (
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.13 // indirect
 	github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.32.1 // indirect
 	github.com/aws/smithy-go v1.20.3 // indirect
-	github.com/benbjohnson/clock v1.3.5 // indirect
 	github.com/danieljoos/wincred v1.2.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/hashicorp/cronexpr v1.1.2 // indirect
 	github.com/hexops/gotextdiff v1.0.3 // indirect
 	github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect

backend/controller/observability/fsm.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/TBD54566975/ftl/backend/schema"
-	"github.com/TBD54566975/ftl/internal/observability"
+
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/metric/noop"
+
+	"github.com/TBD54566975/ftl/backend/schema"
+	"github.com/TBD54566975/ftl/internal/observability"
 )
 
 const (

backend/controller/pubsub/integration_test.go

@@ -65,13 +65,13 @@ func TestConsumptionDelay(t *testing.T) {
 				SELECT created_at, ROW_NUMBER() OVER (ORDER BY created_at) AS row_num
 				FROM (
 				select * from topic_events order by created_at
-				)
+				) AS sub_event_times
 			),
 			async_call_times AS (
 				SELECT created_at, ROW_NUMBER() OVER (ORDER BY created_at) AS row_num
 				FROM (
 				select * from async_calls ac order by created_at
-				)
+				) AS sub_async_calls
 			)
 			SELECT COUNT(*)
 			FROM event_times

backend/controller/pubsub/manager.go

@@ -4,10 +4,11 @@ import (
 	"context"
 	"time"
 
+	"github.com/jpillora/backoff"
+
 	"github.com/TBD54566975/ftl/backend/controller/dal"
 	"github.com/TBD54566975/ftl/backend/controller/scheduledtask"
 	"github.com/TBD54566975/ftl/internal/log"
-	"github.com/jpillora/backoff"
 )
 
 const (