feat: endpoint to query open skyway attestations (#1289)

# Related Github tickets

- VolumeFi#2160

# Background

Create an endpoint to return the blocks of missing skyway events for
validators. This is used by pigeons to know which particular blocks they
should query for events.

# Testing completed

- [x] test coverage exists or has been added/updated
- [x] tested in a private testnet

# Breaking changes

- [x] I have checked my code for breaking changes
- [x] If there are breaking changes, there is a supporting migration.

proto/palomachain/paloma/skyway/query.proto

@@ -116,6 +116,12 @@ service Query {
     option (google.api.http).get =
       "/palomachain/paloma/skyway/pending-batch-for-gas-estimation/{chain_reference_id}";
   }
+
+  rpc GetUnobservedBlocksByAddr(QueryUnobservedBlocksByAddrRequest)
+      returns (QueryUnobservedBlocksByAddrResponse) {
+    option (google.api.http).get =
+      "/palomachain/paloma/skyway/unobserved-blocks-by-addr/{chain_reference_id}";
+  }
 }
 
 message QueryParamsRequest {}
@@ -249,3 +255,12 @@ message QueryLastPendingBatchForGasEstimationRequest {
 message QueryLastPendingBatchForGasEstimationResponse {
   repeated OutgoingTxBatch batch = 1 [ (gogoproto.nullable) = false ];
 }
+
+message QueryUnobservedBlocksByAddrRequest {
+  string chain_reference_id = 1;
+  string address = 2;
+}
+
+message QueryUnobservedBlocksByAddrResponse {
+  repeated uint64 blocks = 1;
+}

x/skyway/keeper/attestation.go

@@ -3,6 +3,7 @@ package keeper
 import (
 	"context"
 	"fmt"
+	"slices"
 	"sort"
 	"strconv"
 
@@ -568,3 +569,53 @@ func (k Keeper) setLatestCompassID(
 	store := k.GetStore(ctx, chainReferenceID)
 	store.Set(types.LatestCompassIDKey, []byte(compassID))
 }
+
+// UnobservedBlocksByAddr returns the blocks with reported events that
+// are still unobserved and are not yet voted by `valAddr`
+func (k Keeper) UnobservedBlocksByAddr(
+	ctx context.Context,
+	chainReferenceID string,
+	valAddr string,
+) ([]uint64, error) {
+	lastCompassID := k.GetLatestCompassID(ctx, chainReferenceID)
+
+	var err error
+	var blocks []uint64
+
+	iterErr := k.IterateAttestations(ctx, chainReferenceID, false, func(_ []byte, att types.Attestation) bool {
+		if att.Observed {
+			return false
+		}
+
+		if slices.Contains(att.Votes, valAddr) {
+			return false
+		}
+
+		var claim types.EthereumClaim
+		claim, err = k.UnpackAttestationClaim(&att)
+		if err != nil {
+			return true
+		}
+
+		// Only include claims from current compass deployment
+		if lastCompassID != "" && claim.GetCompassID() != lastCompassID {
+			return false
+		}
+
+		blocks = append(blocks, claim.GetEthBlockHeight())
+
+		return false
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	if iterErr != nil {
+		return nil, iterErr
+	}
+
+	// Return the blocks in ascending order
+	slices.Sort(blocks)
+
+	return blocks, nil
+}

x/skyway/keeper/grpc_query.go

@@ -428,3 +428,17 @@ func (k Keeper) LastPendingBatchForGasEstimation(ctx context.Context, req *types
 
 	return &types.QueryLastPendingBatchForGasEstimationResponse{Batch: nil}, nil
 }
+
+func (k Keeper) GetUnobservedBlocksByAddr(
+	ctx context.Context,
+	req *types.QueryUnobservedBlocksByAddrRequest,
+) (*types.QueryUnobservedBlocksByAddrResponse, error) {
+	blocks, err := k.UnobservedBlocksByAddr(ctx, req.ChainReferenceId, req.Address)
+	if err != nil {
+		return nil, err
+	}
+
+	return &types.QueryUnobservedBlocksByAddrResponse{
+		Blocks: blocks,
+	}, nil
+}

x/skyway/keeper/grpc_query_test.go

@@ -2,12 +2,16 @@ package keeper
 
 import (
 	"bytes"
+	"context"
 	"math/big"
 	"testing"
 	"time"
 
 	"cosmossdk.io/math"
+	codectypes "github.com/cosmos/cosmos-sdk/codec/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	sdktypes "github.com/cosmos/cosmos-sdk/types"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/palomachain/paloma/x/skyway/types"
 	vtypes "github.com/palomachain/paloma/x/valset/types"
 	"github.com/stretchr/testify/assert"
@@ -538,3 +542,168 @@ func TestQueryPendingSendToRemote(t *testing.T) {
 
 	assert.Equal(t, &expectedRes, response, "json is equal")
 }
+
+func TestGetUnobservedBlocksByAddr(t *testing.T) {
+	chainReferenceID := "test-chain"
+	address := common.BytesToAddress(bytes.Repeat([]byte{0x1}, 20)).String()
+
+	tests := []struct {
+		name     string
+		setup    func(context.Context, Keeper)
+		expected []uint64
+	}{
+		{
+			name: "three unobserved messages for current compass",
+			setup: func(ctx context.Context, k Keeper) {
+				sdkCtx := sdktypes.UnwrapSDKContext(ctx)
+
+				for i := 0; i < 3; i++ {
+					msg := types.MsgLightNodeSaleClaim{
+						SkywayNonce:    uint64(i + 1),
+						EthBlockHeight: uint64(i + 1),
+					}
+					claim, err := codectypes.NewAnyWithValue(&msg)
+					require.NoError(t, err)
+
+					hash, err := msg.ClaimHash()
+					require.NoError(t, err)
+
+					att := &types.Attestation{
+						Observed: false,
+						Votes:    []string{},
+						Height:   uint64(sdkCtx.BlockHeight()),
+						Claim:    claim,
+					}
+
+					k.SetAttestation(ctx, chainReferenceID, uint64(i+1), hash, att)
+				}
+			},
+			expected: []uint64{1, 2, 3},
+		},
+		{
+			name: "three observed messages for current compass",
+			setup: func(ctx context.Context, k Keeper) {
+				sdkCtx := sdktypes.UnwrapSDKContext(ctx)
+
+				for i := 0; i < 3; i++ {
+					msg := types.MsgLightNodeSaleClaim{
+						SkywayNonce:    uint64(i + 1),
+						EthBlockHeight: uint64(i + 1),
+					}
+					claim, err := codectypes.NewAnyWithValue(&msg)
+					require.NoError(t, err)
+
+					hash, err := msg.ClaimHash()
+					require.NoError(t, err)
+
+					att := &types.Attestation{
+						Observed: true,
+						Votes:    []string{},
+						Height:   uint64(sdkCtx.BlockHeight()),
+						Claim:    claim,
+					}
+
+					k.SetAttestation(ctx, chainReferenceID, uint64(i+1), hash, att)
+				}
+			},
+			expected: nil,
+		},
+		{
+			name: "unobserved message already signed by validator",
+			setup: func(ctx context.Context, k Keeper) {
+				sdkCtx := sdktypes.UnwrapSDKContext(ctx)
+
+				for i := 0; i < 3; i++ {
+					msg := types.MsgLightNodeSaleClaim{
+						SkywayNonce:    uint64(i + 1),
+						EthBlockHeight: uint64(i + 1),
+					}
+					claim, err := codectypes.NewAnyWithValue(&msg)
+					require.NoError(t, err)
+
+					hash, err := msg.ClaimHash()
+					require.NoError(t, err)
+
+					att := &types.Attestation{
+						Observed: true,
+						Votes:    []string{address},
+						Height:   uint64(sdkCtx.BlockHeight()),
+						Claim:    claim,
+					}
+
+					k.SetAttestation(ctx, chainReferenceID, uint64(i+1), hash, att)
+				}
+			},
+			expected: nil,
+		},
+		{
+			name: "unobserved messages for current and porevious compass",
+			setup: func(ctx context.Context, k Keeper) {
+				sdkCtx := sdktypes.UnwrapSDKContext(ctx)
+
+				k.setLatestCompassID(ctx, chainReferenceID, "2")
+
+				msg := types.MsgLightNodeSaleClaim{
+					SkywayNonce:    2,
+					EthBlockHeight: 2,
+					CompassId:      "1",
+				}
+				claim, err := codectypes.NewAnyWithValue(&msg)
+				require.NoError(t, err)
+
+				hash, err := msg.ClaimHash()
+				require.NoError(t, err)
+
+				att := &types.Attestation{
+					Observed: false,
+					Votes:    []string{},
+					Height:   uint64(sdkCtx.BlockHeight()),
+					Claim:    claim,
+				}
+
+				k.SetAttestation(ctx, chainReferenceID, 2, hash, att)
+
+				msg = types.MsgLightNodeSaleClaim{
+					SkywayNonce:    1,
+					EthBlockHeight: 1,
+					CompassId:      "2",
+				}
+				claim, err = codectypes.NewAnyWithValue(&msg)
+				require.NoError(t, err)
+
+				hash, err = msg.ClaimHash()
+				require.NoError(t, err)
+
+				att = &types.Attestation{
+					Observed: false,
+					Votes:    []string{},
+					Height:   uint64(sdkCtx.BlockHeight()),
+					Claim:    claim,
+				}
+
+				k.SetAttestation(ctx, chainReferenceID, 1, hash, att)
+			},
+			expected: []uint64{1},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			input := CreateTestEnv(t)
+
+			k := input.SkywayKeeper
+			ctx := input.Context
+
+			tt.setup(ctx, k)
+
+			req := &types.QueryUnobservedBlocksByAddrRequest{
+				ChainReferenceId: chainReferenceID,
+				Address:          address,
+			}
+
+			res, err := k.GetUnobservedBlocksByAddr(ctx, req)
+			require.NoError(t, err)
+			require.Equal(t, tt.expected, res.Blocks)
+		})
+	}
+}