Go SDK: check Go naming conventions with staticcheck

go-sdk/.pre-commit-config.yaml

@@ -66,3 +66,10 @@ repos:
         additional_dependencies: [github.com/segmentio/golines@latest, mvdan.cc/gofumpt@v0.8.0]
         types: [go]
         language: golang
+      - id: staticcheck
+        name: Check naming in Go code
+        entry: staticcheck -checks ST1003 ./...
+        additional_dependencies: [honnef.co/go/tools/cmd/staticcheck@v0.7.0]
+        types: [go]
+        pass_filenames: false
+        language: golang

go-sdk/adr/0007-mixed-lang-task-handler-interface.md

@@ -31,7 +31,7 @@ Proposed.
    `main` reads build, register, serve, with `bundle.Serve()` as its last statement.
    It replaces `BundleProvider` and `Registry`, the callback and the write half of the same bundle.
 2. **`bundle.Register(items ...airflow.Registerable)`** is the single registration verb, taking native Dags and task handlers.
-3. **A Go bundle registers task handlers, not Dags**: `airflow.TaskHandler(dagId, taskId, fn)`, the Go body for a task Python declares with `@task.stub`.
+3. **A Go bundle registers task handlers, not Dags**: `airflow.TaskHandler(dagID, taskID, fn)`, the Go body for a task Python declares with `@task.stub`.
 4. **Both dag_id and task_id are written out on TaskHandler definition**, because Python owns them; nothing is derived from the Go function name.
 5. **Every handler must take an `airflow.Context` first**: a struct embedding `context.Context`, exposing `Logger()`, `Client()`, `TaskInstance()`, and `DagRun()`.
    What Airflow supplies a task arrives as a method on that value rather than as a parameter of its own.
@@ -134,7 +134,7 @@ func (b *BundleRef) Serve() error
 // once there is one.
 type Registerable interface{ registerable() }
 
-func TaskHandler(dagId, taskId string, fn any) Registerable
+func TaskHandler(dagID, taskID string, fn any) Registerable
 
 // Context is what every handler takes first. It is a context, so it passes straight to the
 // logger and the client rather than being stored inside either of them.
@@ -161,7 +161,7 @@ func FromContext(ctx context.Context) (Context, bool)
 
 ## Alternatives
 
-- **`airflow.TaskHandler(dagId, fn, airflow.WithTaskId(...))`**, defaulting the task_id to the Go
+- **`airflow.TaskHandler(dagID, fn, airflow.WithTaskId(...))`**, defaulting the task_id to the Go
   function name. Rejected: see the ids in Context above.
 - **Package-level accessors over a plain `context.Context`** (`airflow.Logger(ctx)`, `airflow.Client(ctx)`), leaving the handler's first parameter as `context.Context`.
   Rejected: `airflow.Logger(ctx)` reads oddly next to `actx.Logger()`, asking the package for something the context already holds.

go-sdk/airflow/bundle.go

@@ -87,7 +87,7 @@ func (b *BundleRef) Register(items ...Registerable) {
 	for _, item := range items {
 		switch item := item.(type) {
 		case *taskHandler:
-			b.taskHandlers.add(item.dagId, item.taskId, item.task)
+			b.taskHandlers.add(item.dagID, item.taskID, item.task)
 		case *DagRef:
 			if item == nil {
 				panic("airflow.BundleRef.Register: cannot register a nil *airflow.DagRef")
@@ -114,32 +114,32 @@ var (
 	_ bundle.EnumerableBundle = (*taskHandlerMap)(nil)
 )
 
-func (m *taskHandlerMap) add(dagId, taskId string, task bundle.Task) {
+func (m *taskHandlerMap) add(dagID, taskID string, task bundle.Task) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
 	if m.handlers == nil {
 		m.handlers = make(map[string]map[string]bundle.Task)
 	}
-	dagHandlers, exists := m.handlers[dagId]
+	dagHandlers, exists := m.handlers[dagID]
 	if !exists {
 		dagHandlers = make(map[string]bundle.Task)
-		m.handlers[dagId] = dagHandlers
+		m.handlers[dagID] = dagHandlers
 	}
-	if _, exists := dagHandlers[taskId]; exists {
+	if _, exists := dagHandlers[taskID]; exists {
 		panic(fmt.Sprintf(
-			"airflow.BundleRef.Register: task %q of Dag %q is already registered", taskId, dagId,
+			"airflow.BundleRef.Register: task %q of Dag %q is already registered", taskID, dagID,
 		))
 	}
-	dagHandlers[taskId] = task
-	m.order = append(m.order, bundle.TaskHandlerInfo{DagID: dagId, TaskID: taskId})
+	dagHandlers[taskID] = task
+	m.order = append(m.order, bundle.TaskHandlerInfo{DagID: dagID, TaskID: taskID})
 }
 
-func (m *taskHandlerMap) LookupTask(dagId, taskId string) (bundle.Task, bool) {
+func (m *taskHandlerMap) LookupTask(dagID, taskID string) (bundle.Task, bool) {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
 
-	task, exists := m.handlers[dagId][taskId]
+	task, exists := m.handlers[dagID][taskID]
 	return task, exists
 }
 

go-sdk/airflow/task_handler.go

@@ -25,7 +25,7 @@ import (
 )
 
 type taskHandler struct {
-	dagId, taskId string
+	dagID, taskID string
 	task          bundle.Task
 }
 
@@ -34,7 +34,7 @@ func (*taskHandler) registerable() {}
 // TaskHandler makes fn the Go body of a task that a Python Dag declares with @task.stub.
 // Pass what it returns to [BundleRef.Register].
 //
-// dagId is the dag_id of that Python Dag, and taskId is the task_id of the stub task.
+// dagID is the dag_id of that Python Dag, and taskID is the task_id of the stub task.
 //
 // fn takes a [Context] first, as the package documentation describes. Every parameter after
 // the Context is data, filled from the arguments of the Python stub's TaskFlow call.
@@ -45,12 +45,12 @@ func (*taskHandler) registerable() {}
 // not a function, does not take a Context first, or does not return an error.
 // main calls TaskHandler before Serve, so a handler that fails the check stops the executable as
 // soon as it starts instead of when the task first runs.
-func TaskHandler(dagId, taskId string, fn any) Registerable {
+func TaskHandler(dagID, taskID string, fn any) Registerable {
 	task, err := newTaskFunction(fn)
 	if err != nil {
-		panic(fmt.Sprintf("airflow.TaskHandler(%q, %q): %v", dagId, taskId, err))
+		panic(fmt.Sprintf("airflow.TaskHandler(%q, %q): %v", dagID, taskID, err))
 	}
-	return &taskHandler{dagId: dagId, taskId: taskId, task: task}
+	return &taskHandler{dagID: dagID, taskID: taskID, task: task}
 }
 
 // newTaskFunction checks fn before bundle.NewTaskFunction does, so that the error names the Go

go-sdk/airflow/task_handler_test.go

@@ -42,15 +42,15 @@ func literalArg(name, jsonType string, value any) map[string]any {
 }
 
 func startupDetails(
-	dagId, taskId string,
+	dagID, taskID string,
 	args ...genmodels.TaskArgBinding,
 ) *genmodels.StartupDetails {
 	mapIndex := -1
 	details := &genmodels.StartupDetails{
 		TI: genmodels.TaskInstance{
 			ID:        "550e8400-e29b-41d4-a716-446655440000",
-			DagID:     dagId,
-			TaskID:    taskId,
+			DagID:     dagID,
+			TaskID:    taskID,
 			RunID:     "run1",
 			TryNumber: 2,
 			MapIndex:  &mapIndex,

go-sdk/example/bundle/concurrentxcom/concurrentxcom_test.go

@@ -54,7 +54,7 @@ func (m *mockXComClient) PushXCom(
 
 func (m *mockXComClient) GetXCom(
 	ctx context.Context,
-	dagId, runId, taskId string,
+	dagID, runID, taskID string,
 	mapIndex *int,
 	key string,
 	value any,

go-sdk/internal/bundle/task.go

@@ -40,7 +40,7 @@ type Task interface {
 // Bundle looks up a registered task by dag_id and task_id. The coordinator
 // runtime uses Bundle to find the task the supervisor asked for.
 type Bundle interface {
-	LookupTask(dagId, taskId string) (Task, bool)
+	LookupTask(dagID, taskID string) (Task, bool)
 }
 
 // TaskHandlerInfo identifies a registered task handler by its dag_id and task_id.

go-sdk/pkg/binding/binding_test.go

@@ -752,7 +752,8 @@ func (s *BindingSuite) TestAnalyzeStructValidation() {
 		Bad chan int `arg:"bad"`
 	}
 	type foldedDuplicateArgNames struct {
-		RegionCode  string
+		RegionCode string
+		//lint:ignore ST1003 the test needs a field that differs from RegionCode only by an underscore
 		Region_code string
 	}
 

go-sdk/pkg/execution/client.go

@@ -38,15 +38,15 @@ const (
 	errCodeXComNotFound       = "XCOM_NOT_FOUND"
 )
 
-// translateApiError converts a supervisor *ApiError whose Err field matches
+// translateAPIError converts a supervisor *APIError whose Err field matches
 // code into a sentinel-wrapped error. Any other error - including a
-// *ApiError with a different code - is returned unchanged so callers can keep
+// *APIError with a different code - is returned unchanged so callers can keep
 // distinguishing transport / server errors from "thing not found".
-func translateApiError(err error, code string, sentinel error, key string) error {
+func translateAPIError(err error, code string, sentinel error, key string) error {
 	if err == nil {
 		return nil
 	}
-	var apiErr *ApiError
+	var apiErr *APIError
 	if errors.As(err, &apiErr) && apiErr.Err == code {
 		return fmt.Errorf("%w: %q", sentinel, key)
 	}
@@ -79,7 +79,7 @@ func (c *CoordinatorClient) GetVariable(ctx context.Context, key string) (string
 		genmodels.GetVariable{Key: key},
 	)
 	if err != nil {
-		return "", translateApiError(err, errCodeVariableNotFound, sdk.VariableNotFound, key)
+		return "", translateAPIError(err, errCodeVariableNotFound, sdk.VariableNotFound, key)
 	}
 
 	var result genmodels.VariableResult
@@ -154,7 +154,7 @@ func (c *CoordinatorClient) GetConnection(
 		genmodels.GetConnection{ConnID: connID},
 	)
 	if err != nil {
-		return sdk.Connection{}, translateApiError(
+		return sdk.Connection{}, translateAPIError(
 			err, errCodeConnectionNotFound, sdk.ConnectionNotFound, connID,
 		)
 	}
@@ -197,16 +197,16 @@ func (c *CoordinatorClient) GetConnection(
 // GetXCom requests an XCom value from the supervisor.
 func (c *CoordinatorClient) GetXCom(
 	ctx context.Context,
-	dagId, runId, taskId string,
+	dagID, runID, taskID string,
 	mapIndex *int,
 	key string,
 	_ any,
 ) (any, error) {
 	msg := genmodels.GetXCom{
 		Key:    key,
-		DagID:  dagId,
-		TaskID: taskId,
-		RunID:  runId,
+		DagID:  dagID,
+		TaskID: taskID,
+		RunID:  runID,
 	}
 	// Assign the pointer, not the dereferenced int: map_index is a nullable
 	// interface{} field and msgpack's omitempty treats an interface{} holding
@@ -217,7 +217,7 @@ func (c *CoordinatorClient) GetXCom(
 
 	resp, err := c.comm.Communicate(ctx, msg)
 	if err != nil {
-		return nil, translateApiError(err, errCodeXComNotFound, sdk.XComNotFound, key)
+		return nil, translateAPIError(err, errCodeXComNotFound, sdk.XComNotFound, key)
 	}
 
 	var result genmodels.XComResult

go-sdk/pkg/execution/client_test.go

@@ -137,7 +137,7 @@ func TestCoordinatorClientErrorTranslation(t *testing.T) {
 	}
 }
 
-// TestCoordinatorClientErrorPassThrough verifies that unrelated *ApiError
+// TestCoordinatorClientErrorPassThrough verifies that unrelated *APIError
 // values (e.g. a generic API_SERVER_ERROR) are returned unchanged.
 func TestCoordinatorClientErrorPassThrough(t *testing.T) {
 	responsePayload := encodeResponseFrame(t, 0, nil, map[string]any{
@@ -156,7 +156,7 @@ func TestCoordinatorClientErrorPassThrough(t *testing.T) {
 	require.Error(t, err)
 	assert.False(t, errors.Is(err, sdk.VariableNotFound),
 		"generic supervisor errors must not be translated to VariableNotFound")
-	var apiErr *ApiError
+	var apiErr *APIError
 	require.True(t, errors.As(err, &apiErr))
 	assert.Equal(t, "API_SERVER_ERROR", apiErr.Err)
 }
@@ -267,7 +267,7 @@ func TestCoordinatorClientVariableWriteErrors(t *testing.T) {
 			comm := NewCoordinatorComm(&responseBuf, io.Discard, logger)
 			client := NewCoordinatorClient(comm)
 
-			var apiErr *ApiError
+			var apiErr *APIError
 			require.ErrorAs(t, tc.call(client), &apiErr)
 			assert.Equal(t, "API_SERVER_ERROR", apiErr.Err)
 		})

go-sdk/pkg/execution/comms.go

@@ -121,7 +121,7 @@ func (c *CoordinatorComm) SendRequest(id int64, body any) error {
 //
 // If the response carries an error (either as the third element of a 3-tuple
 // frame or as a body whose "type" is "ErrorResponse") it is returned as an
-// *ApiError. If the dispatcher's read loop has terminated, the underlying read
+// *APIError. If the dispatcher's read loop has terminated, the underlying read
 // error is returned wrapped in ErrDispatcherClosed.
 //
 // On success it returns the response body's raw msgpack bytes; the caller
@@ -248,15 +248,20 @@ func (c *CoordinatorComm) readLoop() {
 	}
 }
 
-// ApiError represents an error returned by the supervisor over the comm socket.
-type ApiError struct {
+// APIError represents an error returned by the supervisor over the comm socket.
+type APIError struct {
 	Err    string
 	Detail any
 }
 
-func (e *ApiError) Error() string {
+func (e *APIError) Error() string {
 	if e.Detail != nil {
 		return fmt.Sprintf("[%s] %v", e.Err, e.Detail)
 	}
 	return e.Err
 }
+
+// Deprecated: Use APIError.
+//
+//lint:ignore ST1003 kept so code that type-asserts *execution.ApiError still compiles
+type ApiError = APIError

go-sdk/pkg/execution/comms_test.go

@@ -130,7 +130,7 @@ func TestCoordinatorCommCommunicateError(t *testing.T) {
 	)
 	require.Error(t, err)
 
-	apiErr, ok := err.(*ApiError)
+	apiErr, ok := err.(*APIError)
 	require.True(t, ok)
 	assert.Equal(t, "not_found", apiErr.Err)
 }
@@ -158,16 +158,16 @@ func TestCoordinatorCommCommunicateBodyError(t *testing.T) {
 	)
 	require.Error(t, err)
 
-	apiErr, ok := err.(*ApiError)
+	apiErr, ok := err.(*APIError)
 	require.True(t, ok)
 	assert.Equal(t, "server_error", apiErr.Err)
 }
 
-func TestApiErrorFormat(t *testing.T) {
-	err := &ApiError{Err: "not_found", Detail: "Variable 'x' not found"}
+func TestAPIErrorFormat(t *testing.T) {
+	err := &APIError{Err: "not_found", Detail: "Variable 'x' not found"}
 	assert.Equal(t, "[not_found] Variable 'x' not found", err.Error())
 
-	err2 := &ApiError{Err: "server_error"}
+	err2 := &APIError{Err: "server_error"}
 	assert.Equal(t, "server_error", err2.Error())
 }
 

go-sdk/pkg/execution/integration_test.go

@@ -83,21 +83,21 @@ type testBundle map[string]testDag
 
 type testDag map[string]bundle.Task
 
-func (b testBundle) AddDag(dagId string) testDag {
-	b[dagId] = testDag{}
-	return b[dagId]
+func (b testBundle) AddDag(dagID string) testDag {
+	b[dagID] = testDag{}
+	return b[dagID]
 }
 
-func (d testDag) AddTaskWithName(taskId string, fn any) {
+func (d testDag) AddTaskWithName(taskID string, fn any) {
 	task, err := bundle.NewTaskFunction(fn)
 	if err != nil {
 		panic(err)
 	}
-	d[taskId] = task
+	d[taskID] = task
 }
 
-func (b testBundle) LookupTask(dagId, taskId string) (bundle.Task, bool) {
-	task, ok := b[dagId][taskId]
+func (b testBundle) LookupTask(dagID, taskID string) (bundle.Task, bool) {
+	task, ok := b[dagID][taskID]
 	return task, ok
 }
 

go-sdk/pkg/execution/messages.go

@@ -123,7 +123,7 @@ func decodeBody(raw msgpack.RawMessage, dst any) error {
 // apiErrorFromFrame returns the supervisor error carried by a frame, or nil if
 // it is not an error reply. An error arrives either as the third element of a
 // 3-tuple frame (frame.Err) or as a 2-tuple body whose "type" is "ErrorResponse".
-func apiErrorFromFrame(f IncomingFrame) *ApiError {
+func apiErrorFromFrame(f IncomingFrame) *APIError {
 	var raw msgpack.RawMessage
 	switch {
 	case !isNilRaw(f.Err):
@@ -146,7 +146,7 @@ func apiErrorFromFrame(f IncomingFrame) *ApiError {
 		if errCode == "" {
 			errCode = string(genmodels.ErrorTypeGENERICERROR)
 		}
-		return &ApiError{
+		return &APIError{
 			Err:    errCode,
 			Detail: fmt.Sprintf("undecodable error frame: %v", err),
 		}
@@ -155,7 +155,7 @@ func apiErrorFromFrame(f IncomingFrame) *ApiError {
 	if resp.Detail != nil {
 		detail = map[string]any(*resp.Detail)
 	}
-	return &ApiError{Err: string(resp.Error), Detail: detail}
+	return &APIError{Err: string(resp.Error), Detail: detail}
 }
 
 // ifaceString returns the string carried by a nullable schema field decoded as