feat(classifier): deadline exceeded in peer create as connectivity (#3005)

Author: iamKunalGupta

flow/alerting/classifier.go

@@ -478,6 +478,17 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	var peerCreateError *exceptions.PeerCreateError
+	if errors.As(err, &peerCreateError) {
+		// Check for context deadline exceeded error
+		if errors.Is(peerCreateError, context.DeadlineExceeded) {
+			return ErrorNotifyConnectivity, ErrorInfo{
+				Source: ErrorSourceOther,
+				Code:   "CONTEXT_DEADLINE_EXCEEDED",
+			}
+		}
+	}
+
 	return ErrorOther, ErrorInfo{
 		Source: ErrorSourceOther,
 		Code:   "UNKNOWN",

flow/alerting/classifier_test.go

@@ -1,6 +1,7 @@
 package alerting
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"net"
@@ -301,3 +302,14 @@ func TestRandomErrorShouldBeOther(t *testing.T) {
 		Code:   "UNKNOWN",
 	}, errInfo, "Unexpected error info")
 }
+
+func TestPeerCreateTimeoutErrorShouldBeConnectivity(t *testing.T) {
+	// Simulate a peer create timeout error, this is just a unit test, maybe we should try recreating this error in a more realistic way
+	err := exceptions.NewPeerCreateError(context.DeadlineExceeded)
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("error in peer create: %w", err))
+	assert.Equal(t, ErrorNotifyConnectivity, errorClass, "Unexpected error class")
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceOther,
+		Code:   "CONTEXT_DEADLINE_EXCEEDED",
+	}, errInfo, "Unexpected error info")
+}

flow/connectors/core.go

@@ -24,6 +24,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/otel_metrics"
 	"github.com/PeerDB-io/peerdb/flow/shared"
+	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 )
 
 type Connector interface {
@@ -467,7 +468,7 @@ func GetAs[T Connector](ctx context.Context, env map[string]string, config *prot
 	var none T
 	conn, err := GetConnector(ctx, env, config)
 	if err != nil {
-		return none, err
+		return none, exceptions.NewPeerCreateError(err)
 	}
 
 	if tconn, ok := conn.(T); ok {

flow/shared/exceptions/peer.go

@@ -0,0 +1,17 @@
+package exceptions
+
+type PeerCreateError struct {
+	error
+}
+
+func NewPeerCreateError(err error) *PeerCreateError {
+	return &PeerCreateError{err}
+}
+
+func (e *PeerCreateError) Error() string {
+	return "PeerCreate Error: " + e.error.Error()
+}
+
+func (e *PeerCreateError) Unwrap() error {
+	return e.error
+}