coordinator: validate location before marshaling

UpsertCannonLocation and UpsertRelayMonitorLocation dereferenced the
incoming location without checking for nil, so a request with the
field omitted crashed the whole server. Both handlers now reject a
nil location with InvalidArgument before touching it.

Marshal also accepted a request where the type was set but the
matching data payload was missing, and silently wrote an empty
marker instead of the real value. That marker looks like a valid
location to every reader downstream, so it quietly resets whatever
checkpoint or cursor the entry represented. Both the cannon and
relay monitor Marshal implementations now reject this case with a
new ErrLocationDataRequired error. The cannon implementation also
had several types marshaling inline instead of going through the
shared helper, so those were consolidated to keep the check in one
place.

Added a recovery interceptor to the grpc server so a panic in a
handler returns an internal error to the caller instead of taking
down the process. This is a backstop for bugs like the one above,
not a replacement for validating input.

Author: damilolaedwards

pkg/server/persistence/cannon/location.go

@@ -27,12 +27,21 @@ type Location struct {
 }
 
 var (
-	ErrFailedToMarshal   = errors.New("failed to marshal location")
-	ErrFailedToUnmarshal = errors.New("failed to unmarshal location")
+	ErrFailedToMarshal      = errors.New("failed to marshal location")
+	ErrFailedToUnmarshal    = errors.New("failed to unmarshal location")
+	ErrLocationDataRequired = errors.New("location data is required")
 )
 
 // marshalLocationData protojson-marshals data into l.Value and sets l.Type.
+// data must be a non-nil, populated message: an absent oneof value (Type set
+// but its corresponding Data variant not populated) is rejected rather than
+// silently written as an empty "{}" marker, which would otherwise reset
+// whatever cursor/progress this location represents.
 func (l *Location) marshalLocationData(typ string, data proto.Message) error {
+	if data == nil || !data.ProtoReflect().IsValid() {
+		return fmt.Errorf("%w: type %s", ErrLocationDataRequired, typ)
+	}
+
 	l.Type = typ
 
 	b, err := protojson.Marshal(data)
@@ -60,174 +69,35 @@ func (l *Location) Marshal(msg *xatu.CannonLocation) error {
 
 	switch msg.Type {
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_VOLUNTARY_EXIT:
-		l.Type = "BEACON_API_ETH_V2_BEACON_BLOCK_VOLUNTARY_EXIT"
-
-		data := msg.GetEthV2BeaconBlockVoluntaryExit()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK_VOLUNTARY_EXIT", msg.GetEthV2BeaconBlockVoluntaryExit())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_PROPOSER_SLASHING:
-		l.Type = "BEACON_API_ETH_V2_BEACON_BLOCK_PROPOSER_SLASHING"
-
-		data := msg.GetEthV2BeaconBlockProposerSlashing()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK_PROPOSER_SLASHING", msg.GetEthV2BeaconBlockProposerSlashing())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_DEPOSIT:
-		l.Type = "BEACON_API_ETH_V2_BEACON_BLOCK_DEPOSIT"
-
-		data := msg.GetEthV2BeaconBlockDeposit()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK_DEPOSIT", msg.GetEthV2BeaconBlockDeposit())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_ATTESTER_SLASHING:
-		l.Type = "BEACON_API_ETH_V2_BEACON_BLOCK_ATTESTER_SLASHING"
-
-		data := msg.GetEthV2BeaconBlockAttesterSlashing()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
-
+		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK_ATTESTER_SLASHING", msg.GetEthV2BeaconBlockAttesterSlashing())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_EXECUTION_TRANSACTION:
-		l.Type = "BEACON_API_ETH_V2_BEACON_BLOCK_EXECUTION_TRANSACTION"
-
-		data := msg.GetEthV2BeaconBlockExecutionTransaction()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
-
+		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK_EXECUTION_TRANSACTION", msg.GetEthV2BeaconBlockExecutionTransaction())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_BLS_TO_EXECUTION_CHANGE:
-		l.Type = "BEACON_API_ETH_V2_BEACON_BLOCK_BLS_TO_EXECUTION_CHANGE"
-
-		data := msg.GetEthV2BeaconBlockBlsToExecutionChange()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK_BLS_TO_EXECUTION_CHANGE", msg.GetEthV2BeaconBlockBlsToExecutionChange())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_WITHDRAWAL:
-		l.Type = "BEACON_API_ETH_V2_BEACON_BLOCK_WITHDRAWAL"
-
-		data := msg.GetEthV2BeaconBlockWithdrawal()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
-
+		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK_WITHDRAWAL", msg.GetEthV2BeaconBlockWithdrawal())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK:
-		l.Type = "BEACON_API_ETH_V2_BEACON_BLOCK"
-
-		data := msg.GetEthV2BeaconBlock()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
-
+		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK", msg.GetEthV2BeaconBlock())
 	case xatu.CannonType_BEACON_API_ETH_V1_BEACON_BLOB_SIDECAR:
-		l.Type = "BEACON_API_ETH_V1_BEACON_BLOB_SIDECAR"
-
-		data := msg.GetEthV1BeaconBlobSidecar()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V1_BEACON_BLOB_SIDECAR", msg.GetEthV1BeaconBlobSidecar())
 	case xatu.CannonType_BEACON_API_ETH_V1_PROPOSER_DUTY:
-		l.Type = "BEACON_API_ETH_V1_PROPOSER_DUTY"
-
-		data := msg.GetEthV1BeaconProposerDuty()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V1_PROPOSER_DUTY", msg.GetEthV1BeaconProposerDuty())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_ELABORATED_ATTESTATION:
-		l.Type = "BEACON_API_ETH_V2_BEACON_BLOCK_ELABORATED_ATTESTATION"
-
-		data := msg.GetEthV2BeaconBlockElaboratedAttestation()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK_ELABORATED_ATTESTATION", msg.GetEthV2BeaconBlockElaboratedAttestation())
 	case xatu.CannonType_BEACON_API_ETH_V1_BEACON_VALIDATORS:
-		l.Type = "BEACON_API_ETH_V1_BEACON_VALIDATORS"
-
-		data := msg.GetEthV1BeaconValidators()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V1_BEACON_VALIDATORS", msg.GetEthV1BeaconValidators())
 	case xatu.CannonType_BEACON_API_ETH_V1_BEACON_COMMITTEE:
-		l.Type = "BEACON_API_ETH_V1_BEACON_COMMITTEE"
-
-		data := msg.GetEthV1BeaconCommittee()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V1_BEACON_COMMITTEE", msg.GetEthV1BeaconCommittee())
 	case xatu.CannonType_BEACON_API_ETH_V1_BEACON_SYNC_COMMITTEE:
-		l.Type = "BEACON_API_ETH_V1_BEACON_SYNC_COMMITTEE"
-
-		data := msg.GetEthV1BeaconSyncCommittee()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V1_BEACON_SYNC_COMMITTEE", msg.GetEthV1BeaconSyncCommittee())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_SYNC_AGGREGATE:
-		l.Type = "BEACON_API_ETH_V2_BEACON_BLOCK_SYNC_AGGREGATE"
-
-		data := msg.GetEthV2BeaconBlockSyncAggregate()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK_SYNC_AGGREGATE", msg.GetEthV2BeaconBlockSyncAggregate())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_EXECUTION_REQUEST_DEPOSIT:
 		return l.marshalLocationData("BEACON_API_ETH_V2_BEACON_BLOCK_EXECUTION_REQUEST_DEPOSIT", msg.GetEthV2BeaconBlockExecutionRequestDeposit())
 	case xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_EXECUTION_REQUEST_WITHDRAWAL:
@@ -251,27 +121,9 @@ func (l *Location) Marshal(msg *xatu.CannonLocation) error {
 	case xatu.CannonType_BEACON_API_ETH_V1_BEACON_STATE_PENDING_CONSOLIDATION:
 		return l.marshalLocationData("BEACON_API_ETH_V1_BEACON_STATE_PENDING_CONSOLIDATION", msg.GetEthV1BeaconStatePendingConsolidation())
 	case xatu.CannonType_EXECUTION_CANONICAL_BLOCK:
-		l.Type = "EXECUTION_CANONICAL_BLOCK"
-
-		data := msg.GetExecutionCanonicalBlock()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("EXECUTION_CANONICAL_BLOCK", msg.GetExecutionCanonicalBlock())
 	case xatu.CannonType_EXECUTION_CANONICAL_TRANSACTION:
-		l.Type = "EXECUTION_CANONICAL_TRANSACTION"
-
-		data := msg.GetExecutionCanonicalTransaction()
-
-		b, err := protojson.Marshal(data)
-		if err != nil {
-			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-		}
-
-		l.Value = string(b)
+		return l.marshalLocationData("EXECUTION_CANONICAL_TRANSACTION", msg.GetExecutionCanonicalTransaction())
 	case xatu.CannonType_EXECUTION_CANONICAL_LOGS:
 		return l.marshalLocationData("EXECUTION_CANONICAL_LOGS", msg.GetExecutionCanonicalLogs())
 	case xatu.CannonType_EXECUTION_CANONICAL_TRACES:
@@ -303,8 +155,6 @@ func (l *Location) Marshal(msg *xatu.CannonLocation) error {
 	default:
 		return fmt.Errorf("unknown type: %s", msg.Type)
 	}
-
-	return nil
 }
 
 func (l *Location) Unmarshal() (*xatu.CannonLocation, error) {

pkg/server/persistence/cannon/location_marshal_test.go

@@ -0,0 +1,80 @@
+package cannon
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/ethpandaops/xatu/pkg/proto/xatu"
+)
+
+// A request with Type set but the Data oneof left empty is a legal wire
+// message. Marshal used to accept it and silently write an empty JSON
+// object as the value, which reset whatever checkpoint that location
+// represented. It should now reject the request instead.
+func TestMarshal_TypeSetDataAbsent_ReturnsError(t *testing.T) {
+	msg := &xatu.CannonLocation{
+		NetworkId: "mainnet",
+		Type:      xatu.CannonType_BEACON_API_ETH_V2_BEACON_BLOCK_VOLUNTARY_EXIT,
+	}
+
+	l := &Location{}
+
+	err := l.Marshal(msg)
+	if err == nil {
+		t.Fatalf("expected an error, got nil (l.Value = %q)", l.Value)
+	}
+
+	if !errors.Is(err, ErrLocationDataRequired) {
+		t.Fatalf("expected ErrLocationDataRequired, got: %v", err)
+	}
+
+	if l.Value != "" {
+		t.Fatalf("l.Value should be untouched on rejection, got %q", l.Value)
+	}
+}
+
+// Every Type-set/Data-absent case is expected to go through the same
+// marshalLocationData check, so a few representative types are covered
+// explicitly here too.
+func TestMarshal_TypeSetDataAbsent_ReturnsError_AcrossTypes(t *testing.T) {
+	cases := []struct {
+		name string
+		typ  xatu.CannonType
+	}{
+		{"EXECUTION_CANONICAL_BLOCK", xatu.CannonType_EXECUTION_CANONICAL_BLOCK},
+		{"EXECUTION_CANONICAL_TRANSACTION", xatu.CannonType_EXECUTION_CANONICAL_TRANSACTION},
+		{"BEACON_API_ETH_V1_BEACON_STATE_FINALITY_CHECKPOINT", xatu.CannonType_BEACON_API_ETH_V1_BEACON_STATE_FINALITY_CHECKPOINT},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			l := &Location{}
+
+			err := l.Marshal(&xatu.CannonLocation{NetworkId: "mainnet", Type: tc.typ})
+			if !errors.Is(err, ErrLocationDataRequired) {
+				t.Fatalf("expected ErrLocationDataRequired for %s, got: %v", tc.name, err)
+			}
+		})
+	}
+}
+
+// Rows written before this validation existed may still contain an empty
+// marker. Unmarshal should keep reading them back without error, since
+// there is no migration to clean up historical rows.
+func TestUnmarshalOfLegacyEmptyMarker_StillReadsBackSafely(t *testing.T) {
+	l := &Location{
+		NetworkID: "mainnet",
+		Type:      "BEACON_API_ETH_V2_BEACON_BLOCK_VOLUNTARY_EXIT",
+		Value:     "{}",
+	}
+
+	msg, err := l.Unmarshal()
+	if err != nil {
+		t.Fatalf("Unmarshal returned an error reading legacy data: %v", err)
+	}
+
+	data := msg.GetEthV2BeaconBlockVoluntaryExit()
+	if data == nil {
+		t.Fatal("expected a non-nil, zero-valued VoluntaryExit data message")
+	}
+}

pkg/server/persistence/relaymonitor/location.go

@@ -30,8 +30,9 @@ type Location struct {
 }
 
 var (
-	ErrFailedToMarshal   = errors.New("failed to marshal location")
-	ErrFailedToUnmarshal = errors.New("failed to unmarshal location")
+	ErrFailedToMarshal      = errors.New("failed to marshal location")
+	ErrFailedToUnmarshal    = errors.New("failed to unmarshal location")
+	ErrLocationDataRequired = errors.New("location data is required")
 )
 
 // Marshal marshals a proto message into the Location fields.
@@ -45,28 +46,32 @@ func (l *Location) Marshal(msg *xatu.RelayMonitorLocation) error {
 		l.Type = "RELAY_MONITOR_BID_TRACE"
 
 		data := msg.GetBidTrace()
-		if data != nil {
-			b, err := protojson.Marshal(data)
-			if err != nil {
-				return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-			}
+		if data == nil || !data.ProtoReflect().IsValid() {
+			return fmt.Errorf("%w: type %s", ErrLocationDataRequired, l.Type)
+		}
 
-			l.Value = string(b)
+		b, err := protojson.Marshal(data)
+		if err != nil {
+			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
 		}
 
+		l.Value = string(b)
+
 	case xatu.RelayMonitorType_RELAY_MONITOR_PAYLOAD_DELIVERED:
 		l.Type = "RELAY_MONITOR_PAYLOAD_DELIVERED"
 
 		data := msg.GetPayloadDelivered()
-		if data != nil {
-			b, err := protojson.Marshal(data)
-			if err != nil {
-				return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
-			}
+		if data == nil || !data.ProtoReflect().IsValid() {
+			return fmt.Errorf("%w: type %s", ErrLocationDataRequired, l.Type)
+		}
 
-			l.Value = string(b)
+		b, err := protojson.Marshal(data)
+		if err != nil {
+			return fmt.Errorf("%w: %s", ErrFailedToMarshal, err)
 		}
 
+		l.Value = string(b)
+
 	default:
 		return fmt.Errorf("unknown type: %s", msg.Type)
 	}