Enforce API errors via the type system (#3512)

Sadly there doesn't seem to be a way to completely avoid AST code but at
least it is simpler and the consumer side is as pleasant as it can be

Author: ilidemi

.github/actions/genprotos/action.yml

@@ -10,9 +10,10 @@ runs:
       with:
         path: |
           ./flow/generated/protos
+          ./flow/cmd/typed_handler.go
           ./nexus/pt/src/gen
           ./ui/grpc_generated
-        key: ${{ runner.os }}-build-genprotos-${{ hashFiles('buf.gen.yaml', './protos/peers.proto', './protos/flow.proto', './protos/route.proto') }}
+        key: ${{ runner.os }}-build-genprotos-${{ hashFiles('buf.gen.yaml', './protos/peers.proto', './protos/flow.proto', './protos/route.proto', './flow/cmd/gen-grpc-wrapper/main.go') }}
 
     - if: steps.cache.outputs.cache-hit != 'true'
       uses: bufbuild/buf-action@8f4a1456a0ab6a1eb80ba68e53832e6fcfacc16c # v1
@@ -21,4 +22,6 @@ runs:
         github_token: ${{ github.token }}
     - if: steps.cache.outputs.cache-hit != 'true'
       shell: sh
-      run: buf generate protos
+      run: |
+        buf generate protos
+        cd flow && go generate

flow/.gitignore

@@ -10,3 +10,6 @@ vendor/
 
 # goenv local version. See https://github.com/syndbg/goenv/blob/master/COMMANDS.md#goenv-local for more info.
 .go-version
+
+# generated files
+cmd/typed_handler.go

flow/cmd/alerts.go

@@ -9,44 +9,49 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/shared"
-	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 )
 
-func (h *FlowRequestHandler) GetAlertConfigs(ctx context.Context, req *protos.GetAlertConfigsRequest) (*protos.GetAlertConfigsResponse, error) {
+func (h *FlowRequestHandler) GetAlertConfigs(
+	ctx context.Context,
+	req *protos.GetAlertConfigsRequest,
+) (*protos.GetAlertConfigsResponse, APIError) {
 	rows, err := h.pool.Query(ctx, "SELECT id,service_type,service_config,enc_key_id,alert_for_mirrors from peerdb_stats.alerting_config")
 	if err != nil {
-		return nil, exceptions.NewInternalApiError(fmt.Errorf("failed to get alert configs: %w", err))
+		return nil, NewInternalApiError(fmt.Errorf("failed to get alert configs: %w", err))
 	}
 
 	configs, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (*protos.AlertConfig, error) {
 		var serviceConfigPayload []byte
 		var encKeyID string
 		config := &protos.AlertConfig{}
 		if err := row.Scan(&config.Id, &config.ServiceType, &serviceConfigPayload, &encKeyID, &config.AlertForMirrors); err != nil {
-			return nil, exceptions.NewInternalApiError(fmt.Errorf("failed to scan alert config: %w", err))
+			return nil, NewInternalApiError(fmt.Errorf("failed to scan alert config: %w", err))
 		}
 		serviceConfig, err := internal.Decrypt(ctx, encKeyID, serviceConfigPayload)
 		if err != nil {
-			return nil, exceptions.NewInternalApiError(fmt.Errorf("failed to decrypt alert config: %w", err))
+			return nil, NewInternalApiError(fmt.Errorf("failed to decrypt alert config: %w", err))
 		}
 		config.ServiceConfig = string(serviceConfig)
 		return config, nil
 	})
 	if err != nil {
-		return nil, exceptions.NewInternalApiError(fmt.Errorf("failed to collect alert configs: %w", err))
+		return nil, NewInternalApiError(fmt.Errorf("failed to collect alert configs: %w", err))
 	}
 
 	return &protos.GetAlertConfigsResponse{Configs: configs}, nil
 }
 
-func (h *FlowRequestHandler) PostAlertConfig(ctx context.Context, req *protos.PostAlertConfigRequest) (*protos.PostAlertConfigResponse, error) {
+func (h *FlowRequestHandler) PostAlertConfig(
+	ctx context.Context,
+	req *protos.PostAlertConfigRequest,
+) (*protos.PostAlertConfigResponse, APIError) {
 	key, err := internal.PeerDBCurrentEncKey(ctx)
 	if err != nil {
-		return nil, exceptions.NewInternalApiError(fmt.Errorf("failed to get current enc key: %w", err))
+		return nil, NewInternalApiError(fmt.Errorf("failed to get current enc key: %w", err))
 	}
 	serviceConfig, err := key.Encrypt(shared.UnsafeFastStringToReadOnlyBytes(req.Config.ServiceConfig))
 	if err != nil {
-		return nil, exceptions.NewInternalApiError(fmt.Errorf("failed to encrypt alert config: %w", err))
+		return nil, NewInternalApiError(fmt.Errorf("failed to encrypt alert config: %w", err))
 	}
 
 	if req.Config.Id == -1 {
@@ -69,7 +74,7 @@ func (h *FlowRequestHandler) PostAlertConfig(ctx context.Context, req *protos.Po
 			key.ID,
 			req.Config.AlertForMirrors,
 		).Scan(&id); err != nil {
-			return nil, exceptions.NewInternalApiError(fmt.Errorf("failed to insert alert config: %w", err))
+			return nil, NewInternalApiError(fmt.Errorf("failed to insert alert config: %w", err))
 		}
 		return &protos.PostAlertConfigResponse{Id: id}, nil
 	} else if _, err := h.pool.Exec(
@@ -81,17 +86,17 @@ func (h *FlowRequestHandler) PostAlertConfig(ctx context.Context, req *protos.Po
 		req.Config.AlertForMirrors,
 		req.Config.Id,
 	); err != nil {
-		return nil, exceptions.NewInternalApiError(fmt.Errorf("failed to update alert config: %w", err))
+		return nil, NewInternalApiError(fmt.Errorf("failed to update alert config: %w", err))
 	}
 	return &protos.PostAlertConfigResponse{Id: req.Config.Id}, nil
 }
 
 func (h *FlowRequestHandler) DeleteAlertConfig(
 	ctx context.Context,
 	req *protos.DeleteAlertConfigRequest,
-) (*protos.DeleteAlertConfigResponse, error) {
+) (*protos.DeleteAlertConfigResponse, APIError) {
 	if _, err := h.pool.Exec(ctx, "delete from peerdb_stats.alerting_config where id = $1", req.Id); err != nil {
-		return nil, exceptions.NewInternalApiError(fmt.Errorf("failed to delete alert config: %w", err))
+		return nil, NewInternalApiError(fmt.Errorf("failed to delete alert config: %w", err))
 	}
 	return &protos.DeleteAlertConfigResponse{}, nil
 }

flow/cmd/api.go

@@ -259,7 +259,7 @@ func APIMain(ctx context.Context, args *APIServerParams) error {
 		return fmt.Errorf("unable to start scheduler workflow: %w", err)
 	}
 
-	protos.RegisterFlowServiceServer(grpcServer, flowHandler)
+	protos.RegisterFlowServiceServer(grpcServer, NewFlowServiceAdapter(flowHandler))
 	grpc_health_v1.RegisterHealthServer(grpcServer, health.NewServer())
 	reflection.Register(grpcServer)
 

flow/cmd/api_error.go

@@ -0,0 +1,142 @@
+package cmd
+
+import (
+	"errors"
+	"log/slog"
+
+	"github.com/gogo/googleapis/google/rpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/protoadapt"
+)
+
+// APIError is a strongly-typed error that must be a gRPC status error.
+// All handler methods should return this type instead of the generic error interface.
+type APIError interface {
+	error
+	GRPCStatus() *status.Status
+	Code() codes.Code
+}
+
+type apiError struct {
+	status *status.Status
+}
+
+func newAPIError(s *status.Status) *apiError {
+	return &apiError{status: s}
+}
+
+func NewInvalidArgumentApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+	return newAPIError(convertToStatus(codes.InvalidArgument, err, details...))
+}
+
+func NewFailedPreconditionApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+	return newAPIError(convertToStatus(codes.FailedPrecondition, err, details...))
+}
+
+func NewInternalApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+	return newAPIError(convertToStatus(codes.Internal, err, details...))
+}
+
+func NewUnavailableApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+	return newAPIError(convertToStatus(codes.Unavailable, err, details...))
+}
+
+func NewUnimplementedApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+	return newAPIError(convertToStatus(codes.Unimplemented, err, details...))
+}
+
+func NewAlreadyExistsApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+	return newAPIError(convertToStatus(codes.AlreadyExists, err, details...))
+}
+
+func NewNotFoundApiError(err error, details ...*rpc.ErrorInfo) *apiError {
+	return newAPIError(convertToStatus(codes.NotFound, err, details...))
+}
+
+func (e *apiError) Error() string {
+	if e.status == nil {
+		return "unknown error"
+	}
+	return e.status.Err().Error()
+}
+
+func (e *apiError) GRPCStatus() *status.Status {
+	return e.status
+}
+
+func (e *apiError) Code() codes.Code {
+	if e.status == nil {
+		return codes.Unknown
+	}
+	return e.status.Code()
+}
+
+func convertToStatus(code codes.Code, err error, details ...*rpc.ErrorInfo) *status.Status {
+	errorStatus := status.New(code, err.Error())
+	if len(details) == 0 {
+		return errorStatus
+	}
+	convertedDetails := make([]protoadapt.MessageV1, len(details))
+	for i, detail := range details {
+		convertedDetails[i] = detail
+	}
+	richStatus, err := errorStatus.WithDetails(convertedDetails...)
+	if err != nil {
+		// This cannot happen because we control all calls to convertToStatus and only pass code != OK and allow only rpc.ErrorInfo in details
+		slog.Error("Failed to convert to grpc proto error", "error", err) //nolint:sloglint // No context in conversion helper
+		return errorStatus
+	}
+
+	return richStatus
+}
+
+// AsAPIError converts an error to APIError if it's a gRPC status error,
+// otherwise wraps it as an Internal error
+func AsAPIError(err error) APIError {
+	if err == nil {
+		return nil
+	}
+
+	if apiErr, ok := err.(APIError); ok {
+		return apiErr
+	}
+
+	if s, ok := status.FromError(err); ok {
+		return newAPIError(s)
+	}
+
+	return NewInternalApiError(err)
+}
+
+const (
+	ErrorInfoReasonClickHousePeer = "CLICKHOUSE_PEER"
+	ErrorInfoReasonMirror         = "MIRROR"
+)
+
+const (
+	ErrorInfoDomain = "peerdb.io"
+)
+
+const (
+	ErrorMetadataDownstreamErrorCode = "downstreamErrorCode"
+	ErrorMetadataOffendingField      = "offendingField"
+)
+
+func NewClickHousePeerErrorInfo(metadata map[string]string) *rpc.ErrorInfo {
+	return &rpc.ErrorInfo{
+		Reason:   ErrorInfoReasonClickHousePeer,
+		Domain:   ErrorInfoDomain,
+		Metadata: metadata,
+	}
+}
+
+func NewMirrorErrorInfo(metadata map[string]string) *rpc.ErrorInfo {
+	return &rpc.ErrorInfo{
+		Reason:   ErrorInfoReasonMirror,
+		Domain:   ErrorInfoDomain,
+		Metadata: metadata,
+	}
+}
+
+var ErrUnderMaintenance = errors.New("PeerDB is under maintenance. Please retry in a few minutes")