[Proof Module] on-chain claim/proof message distribution validation (#…

…620)


Co-authored-by: Redouane Lakrache <[email protected]>

pkg/client/query/sharedquerier.go

@@ -109,7 +109,7 @@ func (sq *sharedQuerier) GetSessionGracePeriodEndHeight(
 	return shared.GetSessionGracePeriodEndHeight(sharedParams, queryHeight), nil
 }
 
-// GetEarliestClaimCommitHeight returns the earliest block height at which a claim
+// GetEarliestSupplierClaimCommitHeight returns the earliest block height at which a claim
 // for the session that includes queryHeight can be committed for a given supplier.
 //
 // TODO_TECHDEBT(#543): We don't really want to have to query the params for every method call.
@@ -142,7 +142,7 @@ func (sq *sharedQuerier) GetEarliestSupplierClaimCommitHeight(ctx context.Contex
 	), nil
 }
 
-// GetEarliestProofCommitHeight returns the earliest block height at which a proof
+// GetEarliestSupplierProofCommitHeight returns the earliest block height at which a proof
 // for the session that includes queryHeight can be committed for a given supplier.
 //
 // TODO_TECHDEBT(#543): We don't really want to have to query the params for every method call.

pkg/relayer/session/claim.go

@@ -10,6 +10,7 @@ import (
 	"github.com/pokt-network/poktroll/pkg/observable/filter"
 	"github.com/pokt-network/poktroll/pkg/observable/logging"
 	"github.com/pokt-network/poktroll/pkg/relayer"
+	"github.com/pokt-network/poktroll/x/shared"
 )
 
 // createClaims maps over the sessionsToClaimObs observable. For each claim batch, it:
@@ -91,12 +92,19 @@ func (rs *relayerSessionsManager) waitForEarliestCreateClaimsHeight(
 
 	logger := rs.logger.With("session_end_height", sessionEndHeight)
 
-	claimWindowOpenHeight, err := rs.sharedQueryClient.GetClaimWindowOpenHeight(ctx, sessionEndHeight)
+	// TODO_TECHDEBT(#543): We don't really want to have to query the params for every method call.
+	// Once `ModuleParamsClient` is implemented, use its replay observable's `#Last()` method
+	// to get the most recently (asynchronously) observed (and cached) value.
+	// TODO_BLOCKER(@bryanchriswhite,#543): We also don't really want to use the current value of the params. Instead,
+	// we should be using the value that the params had for the session which includes queryHeight.
+	sharedParams, err := rs.sharedQueryClient.GetParams(ctx)
 	if err != nil {
 		failedCreateClaimsSessionsCh <- sessionTrees
 		return nil
 	}
 
+	claimWindowOpenHeight := shared.GetClaimWindowOpenHeight(sharedParams, sessionEndHeight)
+
 	// we wait for claimWindowOpenHeight to be received before proceeding since we need its hash
 	// to know where this servicer's claim submission window opens.
 	logger = logger.With("claim_window_open_height", claimWindowOpenHeight)
@@ -118,19 +126,20 @@ func (rs *relayerSessionsManager) waitForEarliestCreateClaimsHeight(
 	logger = logger.With("claim_window_open_block_hash", fmt.Sprintf("%x", claimsWindowOpenBlock.Hash()))
 	logger.Info().Msg("observed earliest claim commit height offset seed block height")
 
-	// Get the earliestClaimsCommitHeight.
+	// Get the earliest claim commit height for this supplier.
 	supplierAddr := sessionTrees[0].GetSupplierAddress().String()
-	earliestClaimsCommitHeight, err := rs.sharedQueryClient.GetEarliestSupplierClaimCommitHeight(ctx, sessionEndHeight, supplierAddr)
-	if err != nil {
-		failedCreateClaimsSessionsCh <- sessionTrees
-		return nil
-	}
+	earliestSupplierClaimsCommitHeight := shared.GetEarliestSupplierClaimCommitHeight(
+		sharedParams,
+		sessionEndHeight,
+		claimsWindowOpenBlock.Hash(),
+		supplierAddr,
+	)
 
-	logger = logger.With("earliest_claim_commit_height", earliestClaimsCommitHeight)
+	logger = logger.With("earliest_claim_commit_height", earliestSupplierClaimsCommitHeight)
 	logger.Info().Msg("waiting & blocking until the earliest claim commit height for this supplier")
 
-	// Wait for the earliestClaimsCommitHeight to be reached before proceeding.
-	_ = rs.waitForBlock(ctx, earliestClaimsCommitHeight)
+	// Wait for the earliestSupplierClaimsCommitHeight to be reached before proceeding.
+	_ = rs.waitForBlock(ctx, earliestSupplierClaimsCommitHeight)
 
 	logger.Info().Msg("observed earliest claim commit height")
 

pkg/relayer/session/proof.go

@@ -94,15 +94,36 @@ func (rs *relayerSessionsManager) waitForEarliestSubmitProofsHeightAndGeneratePr
 
 	proofWindowOpenHeight := shared.GetProofWindowOpenHeight(sharedParams, sessionEndHeight)
 
-	// we wait for proofWindowOpenHeight to be received before proceeding since we need its hash
+	// we wait for proofWindowOpenHeight to be received before proceeding since we need
+	// its hash to seed the pseudo-random number generator for the proof submission
+	// distribution (i.e. earliestSupplierProofCommitHeight).
 	logger = logger.With("proof_window_open_height", proofWindowOpenHeight)
+	logger.Info().Msg("waiting & blocking until the proof window open height")
+
+	proofsWindowOpenBlock := rs.waitForBlock(ctx, proofWindowOpenHeight)
+
+	// TODO_BLOCKER(@bryanchriswhite, @red0ne): After lean client, there's no
+	// guarantee that all session trees have the same supplier address. Group
+	// session trees by supplier to re-use as much code as possible and still
+	// support batching proofs.
+	//
+	// Get the earliest proof commit height for this supplier.
+	supplierAddr := sessionTrees[0].GetSupplierAddress().String()
+	earliestSupplierProofsCommitHeight := shared.GetEarliestSupplierProofCommitHeight(
+		sharedParams,
+		sessionEndHeight,
+		proofsWindowOpenBlock.Hash(),
+		supplierAddr,
+	)
+
+	logger = logger.With("earliest_supplier_proof_commit_height", earliestSupplierProofsCommitHeight)
 	logger.Info().Msg("waiting & blocking for proof path seed block height")
 
 	// proofWindowOpenHeight - 1 is the block that will have its hash used as the
 	// source of entropy for all the session trees in that batch, waiting for it to
 	// be received before proceeding.
-	proofPathSeedBlock := rs.waitForBlock(ctx, proofWindowOpenHeight-1)
-	_ = rs.waitForBlock(ctx, proofWindowOpenHeight)
+	proofPathSeedBlockHeight := earliestSupplierProofsCommitHeight - 1
+	proofPathSeedBlock := rs.waitForBlock(ctx, proofPathSeedBlockHeight)
 
 	logger = logger.With("proof_path_bock_hash", fmt.Sprintf("%x", proofPathSeedBlock.Hash()))
 	logger.Info().Msg("observed proof path seed block height")
@@ -120,22 +141,12 @@ func (rs *relayerSessionsManager) waitForEarliestSubmitProofsHeightAndGeneratePr
 		failedSubmitProofsSessionsCh,
 	)
 
-	// Get the earliestProofsCommitHeight.
-	supplierAddr := sessionTrees[0].GetSupplierAddress()
-	earliestProofsCommitHeight := shared.GetEarliestSupplierProofCommitHeight(
-		sharedParams,
-		sessionEndHeight,
-		proofPathSeedBlock.Hash(),
-		supplierAddr.String(),
-	)
-
-	logger = logger.With("earliest_proof_commit_height", earliestProofsCommitHeight)
-	logger.Info().Msg("waiting & blocking for earliest proof commit height for this supplier")
+	logger.Info().Msg("waiting & blocking for earliest supplier proof commit height")
 
-	// Wait for the earliestProofsCommitHeight to be reached before proceeding.
-	_ = rs.waitForBlock(ctx, earliestProofsCommitHeight)
+	// Wait for the earliestSupplierProofsCommitHeight to be reached before proceeding.
+	_ = rs.waitForBlock(ctx, earliestSupplierProofsCommitHeight)
 
-	logger.Info().Msg("observed earliest proof commit height")
+	logger.Info().Msg("observed earliest supplier proof commit height")
 
 	// Once the earliest submitProofsHeight has been reached, and all proofs have
 	// been generated, return the sessionTrees that have been successfully proven

pkg/relayer/types.go

@@ -2,6 +2,7 @@ package relayer
 
 import (
 	cosmostypes "github.com/cosmos/cosmos-sdk/types"
+
 	"github.com/pokt-network/poktroll/x/service/types"
 	sessiontypes "github.com/pokt-network/poktroll/x/session/types"
 )

tests/integration/tokenomics/tokenomics_example_test.go

@@ -6,10 +6,11 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/pokt-network/poktroll/cmd/poktrolld/cmd"
-	integration "github.com/pokt-network/poktroll/testutil/integration"
+	"github.com/pokt-network/poktroll/testutil/integration"
 	testutilproof "github.com/pokt-network/poktroll/testutil/proof"
 	prooftypes "github.com/pokt-network/poktroll/x/proof/types"
 	sessiontypes "github.com/pokt-network/poktroll/x/session/types"
+	"github.com/pokt-network/poktroll/x/shared"
 	sharedtypes "github.com/pokt-network/poktroll/x/shared/types"
 	tokenomicstypes "github.com/pokt-network/poktroll/x/tokenomics/types"
 )
@@ -35,6 +36,8 @@ func TestTokenomicsIntegrationExample(t *testing.T) {
 	require.NotNil(t, sharedQueryRes, "unexpected nil params query response")
 	require.EqualValues(t, sharedtypes.DefaultParams(), sharedQueryRes.GetParams())
 
+	sharedParams := sharedQueryRes.GetParams()
+
 	// Prepare a request to update the compute_units_to_tokens_multiplier
 	updateTokenomicsParamMsg := &tokenomicstypes.MsgUpdateParam{
 		Authority: integrationApp.GetAuthority(),
@@ -67,28 +70,31 @@ func TestTokenomicsIntegrationExample(t *testing.T) {
 	// Query the session
 	getSessionRes, err := sessionQueryClient.GetSession(integrationApp.GetSdkCtx(), &getSessionReq)
 	require.NoError(t, err)
-	require.NotNil(t, getSessionRes, "unexpected nil queryResponse")
-	sessionEndHeight := int(getSessionRes.Session.Header.SessionEndBlockHeight)
 
-	// Figure out how many blocks we need to wait until the claim window is open
+	session := getSessionRes.GetSession()
+	require.NotNil(t, session, "unexpected nil queryResponse")
+
+	// Figure out how many blocks we need to wait until the earliest claim commit height
 	// Query and validate the default shared params
-	sharedQueryClient = sharedtypes.NewQueryClient(integrationApp.QueryHelper())
-	sharedParamsReq = sharedtypes.QueryParamsRequest{}
-	sharedQueryRes, err = sharedQueryClient.Params(integrationApp.GetSdkCtx(), &sharedParamsReq)
-	require.NoError(t, err)
-	claimOpenWindowNumBlocks := int(sharedQueryRes.Params.ClaimWindowOpenOffsetBlocks)
+	var claimWindowOpenBlockHash []byte
+	earliestClaimCommitHeight := shared.GetEarliestSupplierClaimCommitHeight(
+		&sharedParams,
+		session.GetHeader().GetSessionEndBlockHeight(),
+		claimWindowOpenBlockHash,
+		integrationApp.DefaultSupplier.GetAddress(),
+	)
 
-	// Need to wait until the claim window is open
-	currentBlockHeight := int(integrationApp.GetSdkCtx().BlockHeight())
-	numBlocksUntilClaimWindowIsOpen := int(sessionEndHeight + claimOpenWindowNumBlocks - currentBlockHeight + 1)
+	// Need to wait until the earliest claim commit height
+	currentBlockHeight := integrationApp.GetSdkCtx().BlockHeight()
+	numBlocksUntilClaimWindowIsOpen := int(earliestClaimCommitHeight - currentBlockHeight)
 	for i := 0; i < numBlocksUntilClaimWindowIsOpen; i++ {
 		integrationApp.NextBlock(t)
 	}
 
 	// Create a new claim
 	createClaimMsg := prooftypes.MsgCreateClaim{
-		SupplierAddress: integrationApp.DefaultSupplier.Address,
-		SessionHeader:   getSessionRes.Session.Header,
+		SupplierAddress: integrationApp.DefaultSupplier.GetAddress(),
+		SessionHeader:   session.GetHeader(),
 		RootHash:        testutilproof.SmstRootWithSum(uint64(1)),
 	}
 

testutil/keeper/tokenomics.go

@@ -217,7 +217,6 @@ func NewTokenomicsModuleKeepers(
 	// Prepare the chain's authority
 	cdc := codec.NewProtoCodec(registry)
 	authority := authtypes.NewModuleAddress(govtypes.ModuleName)
-	t.Logf("authority.String(): %s", authority.String())
 
 	// Construct a real account keeper so that public keys can be queried.
 	addrCodec := addresscodec.NewBech32Codec(app.AccountAddressPrefix)

testutil/testclient/testqueryclients/sharedquerier.go

@@ -75,5 +75,24 @@ func NewTestSharedQueryClient(
 		).
 		AnyTimes()
 
+	sharedQuerier.EXPECT().
+		GetEarliestSupplierProofCommitHeight(gomock.Any(), gomock.Any(), gomock.Any()).
+		DoAndReturn(
+			func(
+				ctx context.Context,
+				sessionEndHeight int64,
+				supplierAddr string,
+			) (int64, error) {
+				sharedParams := sharedtypes.DefaultParams()
+				return shared.GetEarliestSupplierProofCommitHeight(
+					&sharedParams,
+					sessionEndHeight,
+					[]byte{},
+					supplierAddr,
+				), nil
+			},
+		).
+		AnyTimes()
+
 	return sharedQuerier
 }

x/proof/keeper/msg_server_create_claim.go

@@ -64,18 +64,6 @@ func (k msgServer) CreateClaim(
 			"supplier", msg.GetSupplierAddress(),
 		)
 
-	/*
-		TODO_BLOCKER(@bryanchriswhite):
-
-		### Msg distribution validation (depends on sessionRes validation)
-		1. [ ] governance-based earliest block offset
-		2. [ ] pseudo-randomize earliest block offset
-
-		### Claim validation
-		1. [x] sessionRes validation
-		2. [ ] msg distribution validation
-	*/
-
 	logger.Info("validated claim")
 
 	// Assign and upsert claim after all validation.

x/proof/keeper/msg_server_create_claim_test.go

@@ -24,6 +24,9 @@ import (
 var defaultMerkleRoot = testproof.SmstRootWithSum(10)
 
 func TestMsgServer_CreateClaim_Success(t *testing.T) {
+	var claimWindowOpenBlockHash []byte
+	supplierAddr := sample.AccAddress()
+
 	tests := []struct {
 		desc              string
 		getClaimMsgHeight func(
@@ -32,8 +35,15 @@ func TestMsgServer_CreateClaim_Success(t *testing.T) {
 		) int64
 	}{
 		{
-			desc:              "claim message height equals claim window open height",
-			getClaimMsgHeight: shared.GetClaimWindowOpenHeight,
+			desc: "claim message height equals supplier's earliest claim commit height",
+			getClaimMsgHeight: func(sharedParams *sharedtypes.Params, queryHeight int64) int64 {
+				return shared.GetEarliestSupplierClaimCommitHeight(
+					sharedParams,
+					queryHeight,
+					claimWindowOpenBlockHash,
+					supplierAddr,
+				)
+			},
 		},
 		{
 			desc:              "claim message height equals claim window close height",
@@ -57,7 +67,6 @@ func TestMsgServer_CreateClaim_Success(t *testing.T) {
 			sessionStartHeight := blockHeight
 
 			service := &sharedtypes.Service{Id: testServiceId}
-			supplierAddr := sample.AccAddress()
 			appAddr := sample.AccAddress()
 
 			keepers.SetSupplier(ctx, sharedtypes.Supplier{
@@ -125,6 +134,8 @@ func TestMsgServer_CreateClaim_Success(t *testing.T) {
 }
 
 func TestMsgServer_CreateClaim_Error_OutsideOfWindow(t *testing.T) {
+	var claimWindowOpenBlockHash []byte
+
 	// Set block height to 1 so there is a valid session on-chain.
 	blockHeightOpt := keepertest.WithBlockHeight(1)
 	keepers, ctx := keepertest.NewProofModuleKeepers(t, blockHeightOpt)
@@ -170,9 +181,11 @@ func TestMsgServer_CreateClaim_Error_OutsideOfWindow(t *testing.T) {
 		sessionHeader.GetSessionEndBlockHeight(),
 	)
 
-	claimWindowOpenHeight := shared.GetClaimWindowOpenHeight(
+	earliestClaimCommitHeight := shared.GetEarliestSupplierClaimCommitHeight(
 		&sharedParams,
 		sessionHeader.GetSessionEndBlockHeight(),
+		claimWindowOpenBlockHash,
+		supplierAddr,
 	)
 
 	tests := []struct {
@@ -182,13 +195,18 @@ func TestMsgServer_CreateClaim_Error_OutsideOfWindow(t *testing.T) {
 	}{
 		{
 			desc:           "claim message height equals claim window open height minus one",
-			claimMsgHeight: claimWindowOpenHeight - 1,
+			claimMsgHeight: earliestClaimCommitHeight - 1,
 			expectedErr: status.Error(
 				codes.FailedPrecondition,
 				types.ErrProofClaimOutsideOfWindow.Wrapf(
-					"current block height (%d) is less than session claim window open height (%d)",
-					claimWindowOpenHeight-1,
-					shared.GetClaimWindowOpenHeight(&sharedParams, sessionHeader.GetSessionEndBlockHeight()),
+					"current block height (%d) is less than the session's earliest claim commit height (%d)",
+					earliestClaimCommitHeight-1,
+					shared.GetEarliestSupplierClaimCommitHeight(
+						&sharedParams,
+						sessionHeader.GetSessionEndBlockHeight(),
+						claimWindowOpenBlockHash,
+						supplierAddr,
+					),
 				).Error(),
 			),
 		},