[On-chain, Relayminer] chore: add claim msg validation

app/app.go

@@ -575,16 +575,6 @@ func New(
 	)
 	serviceModule := servicemodule.NewAppModule(appCodec, app.ServiceKeeper, app.AccountKeeper, app.BankKeeper)
 
-	app.SupplierKeeper = *suppliermodulekeeper.NewKeeper(
-		appCodec,
-		keys[suppliermoduletypes.StoreKey],
-		keys[suppliermoduletypes.MemStoreKey],
-		app.GetSubspace(suppliermoduletypes.ModuleName),
-
-		app.BankKeeper,
-	)
-	supplierModule := suppliermodule.NewAppModule(appCodec, app.SupplierKeeper, app.AccountKeeper, app.BankKeeper)
-
 	app.GatewayKeeper = *gatewaymodulekeeper.NewKeeper(
 		appCodec,
 		keys[gatewaymoduletypes.StoreKey],
@@ -607,6 +597,28 @@ func New(
 	)
 	applicationModule := applicationmodule.NewAppModule(appCodec, app.ApplicationKeeper, app.AccountKeeper, app.BankKeeper)
 
+	// TODO_TECHDEBT: Evaluate if this NB goes away after we upgrade to cosmos 0.5x
+	// NB: there is a circular dependency between the supplier and session keepers.
+	// Because the keepers are values (as opposed to pointers), they are copied
+	// when passed into their respective module constructor functions. For this
+	// reason, the existing pattern of ignite-generated keeper/module construction
+	// must be broken for these keepers and modules.
+	//
+	// Order of operations:
+	// 1. Construct supplier keeper
+	// 2. Construct session keeper
+	// 3. Provide session keeper to supplier keeper via custom #SupplySessionKeeper method.
+	// 4. Construct supplier module
+	// 5. Construct session module
+	app.SupplierKeeper = *suppliermodulekeeper.NewKeeper(
+		appCodec,
+		keys[suppliermoduletypes.StoreKey],
+		keys[suppliermoduletypes.MemStoreKey],
+		app.GetSubspace(suppliermoduletypes.ModuleName),
+
+		app.BankKeeper,
+	)
+
 	app.SessionKeeper = *sessionmodulekeeper.NewKeeper(
 		appCodec,
 		keys[sessionmoduletypes.StoreKey],
@@ -616,6 +628,10 @@ func New(
 		app.ApplicationKeeper,
 		app.SupplierKeeper,
 	)
+
+	app.SupplierKeeper.SupplySessionKeeper(app.SessionKeeper)
+
+	supplierModule := suppliermodule.NewAppModule(appCodec, app.SupplierKeeper, app.AccountKeeper, app.BankKeeper)
 	sessionModule := sessionmodule.NewAppModule(appCodec, app.SessionKeeper, app.AccountKeeper, app.BankKeeper)
 
 	// this line is used by starport scaffolding # stargate/app/keeperDefinition

e2e/tests/init_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/pokt-network/poktroll/app"
+	"github.com/pokt-network/poktroll/testutil/testclient"
 	apptypes "github.com/pokt-network/poktroll/x/application/types"
 	sessiontypes "github.com/pokt-network/poktroll/x/session/types"
 	sharedtypes "github.com/pokt-network/poktroll/x/shared/types"
@@ -59,9 +60,11 @@ func TestMain(m *testing.M) {
 
 type suite struct {
 	gocuke.TestingT
-	pocketd       *pocketdBin
-	scenarioState map[string]any // temporary state for each scenario
-	cdc           codec.Codec
+	// TODO_TECHDEBT: rename to `poktrolld`.
+	pocketd             *pocketdBin
+	scenarioState       map[string]any // temporary state for each scenario
+	cdc                 codec.Codec
+	supplierQueryClient suppliertypes.QueryClient
 }
 
 func (s *suite) Before() {
@@ -71,6 +74,10 @@ func (s *suite) Before() {
 	s.buildAddrMap()
 	s.buildAppMap()
 	s.buildSupplierMap()
+
+	flagSet := testclient.NewLocalnetFlagSet(s)
+	clientCtx := testclient.NewLocalnetClientCtx(s, flagSet)
+	s.supplierQueryClient = suppliertypes.NewQueryClient(clientCtx)
 }
 
 // TestFeatures runs the e2e tests specified in any .features files in this directory
@@ -79,6 +86,7 @@ func TestFeatures(t *testing.T) {
 	gocuke.NewRunner(t, &suite{}).Path(flagFeaturesPath).Run()
 }
 
+// TODO_TECHDEBT: rename `pocketd` to `poktrolld`.
 func (s *suite) TheUserHasThePocketdBinaryInstalled() {
 	s.TheUserRunsTheCommand("help")
 }

e2e/tests/session.feature

@@ -0,0 +1,10 @@
+Feature: Session Namespace
+
+  Scenario: Supplier completes claim/proof lifecycle for a valid session
+    Given the user has the pocketd binary installed
+    When the supplier "supplier1" has serviced a session with "5" relays for service "svc1" for application "app1"
+    And after the supplier creates a claim for the session for service "svc1" for application "app1"
+    Then the claim created by supplier "supplier1" for service "svc1" for application "app1" should be persisted on-chain
+#  TODO_IMPROVE: ...
+#    And an event should be emitted...
+#  TODO_INCOMPLETE: add step(s) for proof validation.

e2e/tests/session_steps_test.go

@@ -0,0 +1,162 @@
+//go:build e2e
+
+package e2e
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	abci "github.com/cometbft/cometbft/abci/types"
+	"github.com/stretchr/testify/require"
+
+	eventsquery "github.com/pokt-network/poktroll/pkg/client/events_query"
+	"github.com/pokt-network/poktroll/pkg/either"
+	"github.com/pokt-network/poktroll/pkg/observable"
+	"github.com/pokt-network/poktroll/pkg/observable/channel"
+	"github.com/pokt-network/poktroll/testutil/testclient"
+	suppliertypes "github.com/pokt-network/poktroll/x/supplier/types"
+)
+
+const (
+	createClaimTimeoutDuration   = 10 * time.Second
+	eitherEventsReplayBufferSize = 100
+	msgClaimSenderQueryFmt       = "tm.event='Tx' AND message.sender='%s'"
+	testServiceId                = "anvil"
+	eitherEventsBzReplayObsKey   = "eitherEventsBzReplayObsKey"
+	preExistingClaimsKey         = "preExistingClaimsKey"
+)
+
+func (s *suite) AfterTheSupplierCreatesAClaimForTheSessionForServiceForApplication(serviceId, appName string) {
+	var ctx, done = context.WithCancel(context.Background())
+
+	// TODO_CONSIDERATION: if this test suite gets more complex, it might make
+	// sense to refactor this key into a function that takes serviceId and appName
+	// as arguments and returns the key.
+	eitherEventsBzReplayObs := s.scenarioState[eitherEventsBzReplayObsKey].(observable.ReplayObservable[either.Bytes])
+
+	// TODO(#220): refactor to use EventsReplayClient once available.
+	channel.ForEach[either.Bytes](
+		ctx, eitherEventsBzReplayObs,
+		func(_ context.Context, eitherEventBz either.Bytes) {
+			eventBz, err := eitherEventBz.ValueOrError()
+			require.NoError(s, err)
+
+			if strings.Contains(string(eventBz), "jsonrpc") {
+				return
+			}
+
+			// Unmarshal event data into a TxEventResponse object.
+			txEvent := &abci.TxResult{}
+			err = json.Unmarshal(eventBz, txEvent)
+			require.NoError(s, err)
+
+			var found bool
+			for _, event := range txEvent.Result.Events {
+				for _, attribute := range event.Attributes {
+					if attribute.Key == "action" {
+						require.Equal(
+							s, "/pocket.supplier.MsgCreateClaim",
+							attribute.Value,
+						)
+						found = true
+						break
+					}
+				}
+				if found {
+					break
+				}
+			}
+			require.Truef(s, found, "unable to find event action attribute")
+
+			done()
+		},
+	)
+
+	select {
+	case <-ctx.Done():
+	case <-time.After(createClaimTimeoutDuration):
+		s.Fatal("timed out waiting for claim to be created")
+	}
+}
+
+func (s *suite) TheClaimCreatedBySupplierForServiceForApplicationShouldBePersistedOnchain(supplierName, serviceId, appName string) {
+	ctx := context.Background()
+
+	claimsRes, err := s.supplierQueryClient.AllClaims(ctx, &suppliertypes.QueryAllClaimsRequest{
+		Filter: &suppliertypes.QueryAllClaimsRequest_SupplierAddress{
+			SupplierAddress: accNameToAddrMap[supplierName],
+		},
+	})
+	require.NoError(s, err)
+	require.NotNil(s, claimsRes)
+
+	// Assert that the number of claims has increased by one.
+	preExistingClaims := s.scenarioState[preExistingClaimsKey].([]suppliertypes.Claim)
+	require.Len(s, claimsRes.Claim, len(preExistingClaims)+1)
+
+	// TODO_IMPROVE: assert that the root hash of the claim contains the correct
+	// SMST sum. The sum can be retrieved by parsing the last 8 bytes as a
+	// binary-encoded uint64; e.g. something like:
+	// `binary.Uvarint(claim.RootHash[len(claim.RootHash-8):])`
+
+	// TODO_IMPROVE: add assertions about serviceId and appName and/or incorporate
+	// them into the scenarioState key(s).
+
+	claim := claimsRes.Claim[0]
+	require.Equal(s, accNameToAddrMap[supplierName], claim.SupplierAddress)
+}
+
+func (s *suite) TheSupplierHasServicedASessionWithRelaysForServiceForApplication(supplierName, relayCountStr, serviceId, appName string) {
+	ctx := context.Background()
+
+	relayCount, err := strconv.Atoi(relayCountStr)
+	require.NoError(s, err)
+
+	// Query for any existing claims so that we can compensate for them in the
+	// future assertions about changes in on-chain claims.
+	claimsRes, err := s.supplierQueryClient.AllClaims(ctx, &suppliertypes.QueryAllClaimsRequest{})
+	require.NoError(s, err)
+	s.scenarioState[preExistingClaimsKey] = claimsRes.Claim
+
+	// Construct an events query client to listen for tx events from the supplier.
+	msgSenderQuery := fmt.Sprintf(msgClaimSenderQueryFmt, accNameToAddrMap[supplierName])
+
+	// TODO_TECHDEBT(#220): refactor to use EventsReplayClient once available.
+	eventsQueryClient := eventsquery.NewEventsQueryClient(testclient.CometLocalWebsocketURL)
+	eitherEventsBzObs, err := eventsQueryClient.EventsBytes(ctx, msgSenderQuery)
+	require.NoError(s, err)
+
+	eitherEventsBytesObs := observable.Observable[either.Bytes](eitherEventsBzObs)
+	eitherEventsBzRelayObs := channel.ToReplayObservable(ctx, eitherEventsReplayBufferSize, eitherEventsBytesObs)
+	s.scenarioState[eitherEventsBzReplayObsKey] = eitherEventsBzRelayObs
+
+	s.sendRelaysForSession(
+		appName,
+		supplierName,
+		testServiceId,
+		relayCount,
+	)
+}
+
+func (s *suite) sendRelaysForSession(
+	appName string,
+	supplierName string,
+	serviceId string,
+	relayLimit int,
+) {
+	s.TheApplicationIsStakedForService(appName, serviceId)
+	s.TheSupplierIsStakedForService(supplierName, serviceId)
+	s.TheSessionForApplicationAndServiceContainsTheSupplier(appName, serviceId, supplierName)
+
+	// TODO_IMPROVE/TODO_COMMUNITY: hard-code a default set of RPC calls to iterate over for coverage.
+	data := `{"jsonrpc":"2.0","method":"eth_blockNumber","params":[],"id":1}`
+
+	for i := 0; i < relayLimit; i++ {
+		s.TheApplicationSendsTheSupplierARequestForServiceWithData(appName, supplierName, serviceId, data)
+		s.TheApplicationReceivesASuccessfulRelayResponseSignedBy(appName, supplierName)
+	}
+}

pkg/client/tx/client.go

@@ -9,8 +9,8 @@ import (
 	"fmt"
 	"sync"
 
+	"cosmossdk.io/api/tendermint/abci"
 	"cosmossdk.io/depinject"
-	abciTypes "github.com/cometbft/cometbft/abci/types"
 	comettypes "github.com/cometbft/cometbft/types"
 	cosmostypes "github.com/cosmos/cosmos-sdk/types"
 	"go.uber.org/multierr"
@@ -87,11 +87,10 @@ type (
 
 // TxEvent is used to deserialize incoming websocket messages from
 // the transactions subscription.
-type TxEvent struct {
-	// Tx is the binary representation of the tx hash.
-	Tx     []byte            `json:"tx"`
-	Events []abciTypes.Event `json:"events"`
-}
+//
+// TODO_CONSIDERATION: either expose this via an interface and unexport this type,
+// or remove it altogether.
+type TxEvent = abci.TxResult
 
 // NewTxClient attempts to construct a new TxClient using the given dependencies
 // and options.

testutil/keeper/supplier.go

@@ -15,12 +15,16 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
 
-	mocks "github.com/pokt-network/poktroll/testutil/supplier/mocks"
+	"github.com/pokt-network/poktroll/testutil/supplier"
+	"github.com/pokt-network/poktroll/testutil/supplier/mocks"
+	sessiontypes "github.com/pokt-network/poktroll/x/session/types"
 	"github.com/pokt-network/poktroll/x/supplier/keeper"
 	"github.com/pokt-network/poktroll/x/supplier/types"
 )
 
-func SupplierKeeper(t testing.TB) (*keeper.Keeper, sdk.Context) {
+func SupplierKeeper(t testing.TB, sessionByAppAddr supplier.SessionsByAppAddress) (*keeper.Keeper, sdk.Context) {
+	t.Helper()
+
 	storeKey := sdk.NewKVStoreKey(types.StoreKey)
 	memStoreKey := storetypes.NewMemoryStoreKey(types.MemStoreKey)
 
@@ -38,6 +42,36 @@ func SupplierKeeper(t testing.TB) (*keeper.Keeper, sdk.Context) {
 	mockBankKeeper.EXPECT().DelegateCoinsFromAccountToModule(gomock.Any(), gomock.Any(), types.ModuleName, gomock.Any()).AnyTimes()
 	mockBankKeeper.EXPECT().UndelegateCoinsFromModuleToAccount(gomock.Any(), types.ModuleName, gomock.Any(), gomock.Any()).AnyTimes()
 
+	mockSessionKeeper := mocks.NewMockSessionKeeper(ctrl)
+	mockSessionKeeper.EXPECT().
+		GetSession(gomock.AssignableToTypeOf(sdk.Context{}), gomock.Any()).
+		DoAndReturn(
+			func(
+				ctx sdk.Context,
+				req *sessiontypes.QueryGetSessionRequest,
+			) (*sessiontypes.QueryGetSessionResponse, error) {
+				session, ok := sessionByAppAddr[req.GetApplicationAddress()]
+				require.Truef(t, ok, "application address not provided during mock construction: %q", req.ApplicationAddress)
+
+				return &sessiontypes.QueryGetSessionResponse{
+					Session: &sessiontypes.Session{
+						Header: &sessiontypes.SessionHeader{
+							ApplicationAddress:      session.GetApplication().GetAddress(),
+							Service:                 req.GetService(),
+							SessionStartBlockHeight: 1,
+							SessionId:               session.GetSessionId(),
+							SessionEndBlockHeight:   5,
+						},
+						SessionId:           session.GetSessionId(),
+						SessionNumber:       1,
+						NumBlocksPerSession: session.GetNumBlocksPerSession(),
+						Application:         session.GetApplication(),
+						Suppliers:           session.GetSuppliers(),
+					},
+				}, nil
+			},
+		).AnyTimes()
+
 	paramsSubspace := typesparams.NewSubspace(cdc,
 		types.Amino,
 		storeKey,
@@ -52,6 +86,7 @@ func SupplierKeeper(t testing.TB) (*keeper.Keeper, sdk.Context) {
 
 		mockBankKeeper,
 	)
+	k.SupplySessionKeeper(mockSessionKeeper)
 
 	ctx := sdk.NewContext(stateStore, tmproto.Header{}, false, log.NewNopLogger())