flowinfra: preserve flowRetryableError correctly across network

This commit makes it so that `flowinfra.flowRetryableError` type is
correctly preserved across network. Previously, if the error originated
on the remote node, the coordinator node would receive
`errbase.opaqueLeaf` error since the decoder method wasn't registered
for the error, now the error is preserved correctly.

Release note: None

pkg/sql/flowinfra/BUILD.bazel

@@ -43,6 +43,7 @@ go_library(
         "//pkg/util/uuid",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
+        "@com_github_gogo_protobuf//proto",
         "@io_opentelemetry_go_otel//attribute",
     ],
 )

pkg/sql/flowinfra/flow_registry.go

@@ -12,7 +12,6 @@ package flowinfra
 
 import (
 	"context"
-	"fmt"
 	"sync"
 	"time"
 
@@ -25,6 +24,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
+	"github.com/gogo/protobuf/proto"
 )
 
 // errNoInboundStreamConnection is the error propagated through the flow when
@@ -239,8 +239,20 @@ type flowRetryableError struct {
 	cause error
 }
 
-func (e *flowRetryableError) Error() string {
-	return fmt.Sprintf("flow retryable error: %+v", e.cause)
+var _ errors.Wrapper = &flowRetryableError{}
+
+func (e *flowRetryableError) Error() string { return e.cause.Error() }
+func (e *flowRetryableError) Cause() error  { return e.cause }
+func (e *flowRetryableError) Unwrap() error { return e.Cause() }
+
+func decodeFlowRetryableError(
+	_ context.Context, cause error, _ string, _ []string, _ proto.Message,
+) error {
+	return &flowRetryableError{cause: cause}
+}
+
+func init() {
+	errors.RegisterWrapperDecoder(errors.GetTypeKey((*flowRetryableError)(nil)), decodeFlowRetryableError)
 }
 
 // IsFlowRetryableError returns true if an error represents a retryable

pkg/testutils/lint/lint_test.go

@@ -1175,6 +1175,7 @@ func TestLint(t *testing.T) {
 			":!spanconfig/errors.go",
 			":!roachpb/replica_unavailable_error.go",
 			":!roachpb/ambiguous_result_error.go",
+			":!sql/flowinfra/flow_registry.go",
 			":!sql/pgwire/pgerror/constraint_name.go",
 			":!sql/pgwire/pgerror/severity.go",
 			":!sql/pgwire/pgerror/with_candidate_code.go",