Merge pull request #76 from DIMO-Network/update-conversion-logging

Add more conversion error handling

Makefile

@@ -53,7 +53,7 @@ test:
 
 lint:
 	@golangci-lint version
-	@golangci-lint run --timeout=10m
+	@golangci-lint run --timeout=30m
 
 format:
 	@golangci-lint run --fix

go.mod

@@ -5,7 +5,7 @@ go 1.22.0
 require (
 	github.com/DIMO-Network/clickhouse-infra v0.0.2
 	github.com/DIMO-Network/devices-api v1.27.38
-	github.com/DIMO-Network/model-garage v0.2.11
+	github.com/DIMO-Network/model-garage v0.2.15
 	github.com/DIMO-Network/nameindexer v0.0.7
 	github.com/DIMO-Network/shared v0.10.18
 	github.com/docker/go-connections v0.5.0

go.sum

@@ -110,8 +110,8 @@ github.com/DIMO-Network/clickhouse-infra v0.0.2 h1:bKbBWgsZEqQKrXCfc93mag0JALg74
 github.com/DIMO-Network/clickhouse-infra v0.0.2/go.mod h1:NX0pmwYzZj1M9sh1RRhFjhMZ0J9OkuMgXMA9Jad6lX0=
 github.com/DIMO-Network/devices-api v1.27.38 h1:FcPMfsJ1WBWRvVMp0ParmPFzAXI+C+t0kb4YrvUY9Uc=
 github.com/DIMO-Network/devices-api v1.27.38/go.mod h1:XAU/MLIXItrFWZO8/+CGvRfmGMpsTBF70r9qRGq9V3w=
-github.com/DIMO-Network/model-garage v0.2.11 h1:3ofR3McKnngQb6ehApA9u80MfCyhugMiM7zBs+svlVM=
-github.com/DIMO-Network/model-garage v0.2.11/go.mod h1:voOvGWwtAnazsciDVTqjPYDttIkXx0YG4xA+MIjM/8k=
+github.com/DIMO-Network/model-garage v0.2.15 h1:KW9M5CrBA6Kpwj5xX9tiIMd3RRgdZhrTbwZkGKi3R9o=
+github.com/DIMO-Network/model-garage v0.2.15/go.mod h1:voOvGWwtAnazsciDVTqjPYDttIkXx0YG4xA+MIjM/8k=
 github.com/DIMO-Network/nameindexer v0.0.7 h1:aigO9WZGvoGkyu2R92ZXZGT44zB3O0bgLB+A7wbmEsk=
 github.com/DIMO-Network/nameindexer v0.0.7/go.mod h1:3X4sme/KVDNZ4A0YNGt/ssfVgSkSRxJ1RKmQYE/Vtqo=
 github.com/DIMO-Network/shared v0.10.18 h1:NKWqIyo5plORii+hgOpXS2zUdmIHFKZuPqoo5Oaf0gk=

internal/dimovss/dimovss.go

@@ -2,6 +2,7 @@ package dimovss
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 
@@ -91,22 +92,41 @@ func (v *vssProcessor) Process(ctx context.Context, msg *service.Message) (servi
 		// ignore v1.1 messages
 		return nil, nil
 	}
-	signals, retErr := convert.SignalsFromPayload(ctx, v.tokenGetter, msgBytes)
-	if errors.As(retErr, &deviceapi.NotFoundError{}) {
-		// If we do not have an Token for this device we want to drop the message. But we don't want to log an error.
-		v.logger.Trace(fmt.Sprintf("dropping message: %v", retErr))
-		return nil, nil
+	var partialErr *service.Message
+	var retMsgs service.MessageBatch
+	signals, err := convert.SignalsFromPayload(ctx, v.tokenGetter, msgBytes)
+	if err != nil {
+		if errors.As(err, &deviceapi.NotFoundError{}) {
+			// If we do not have an Token for this device we want to drop the message. But we don't want to log an error.
+			v.logger.Trace(fmt.Sprintf("dropping message: %v", err))
+			return nil, nil
+		}
+
+		convertErr := convert.ConversionError{}
+		if !errors.As(err, &convertErr) {
+			return nil, fmt.Errorf("failed to convert signals: %w", err)
+		}
+		// if we have a conversion error we will add a error message with metadata to the batch.
+		// but still return the signals that we could decode.
+		partialErr = msg.Copy()
+		partialErr.SetError(err)
+		data, err := json.Marshal(convertErr)
+		if err == nil {
+			partialErr.SetBytes(data)
+		} else {
+			partialErr.SetBytes(nil)
+		}
+		retMsgs = append(retMsgs, partialErr)
+		signals = convertErr.DecodedSignals
 	}
-	retMsgs := make([]*service.Message, len(signals))
+
 	for i := range signals {
 		sigVals := vss.SignalToSlice(signals[i])
-		retMsgs[i] = msg.Copy()
-		retMsgs[i].SetStructured(sigVals)
-	}
-	if retErr != nil {
-		// still reutrn retMsg since we may have partially converted signals.
-		return retMsgs, fmt.Errorf("failed to convert signals: %w", retErr)
+		msgCpy := msg.Copy()
+		msgCpy.SetStructured(sigVals)
+		retMsgs = append(retMsgs, msgCpy)
 	}
+
 	return retMsgs, nil
 }
 

internal/dimovss/dimovss_test.go

@@ -109,6 +109,15 @@ func TestVSSProcessorProcess(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			batch, err := vssProc.Process(context.Background(), test.msg)
 			if test.expectedErr {
+				if err == nil {
+					return
+				}
+				for _, msg := range batch {
+					err = msg.GetError()
+					if err != nil {
+						break
+					}
+				}
 				require.Error(t, err)
 			} else {
 				require.NoError(t, err)