Repository: codenotary/immudb
File: pkg/integration/replication/synchronous_replication_test.go (Go)

Code before:
func (suite *SyncTestSuite) TestPrecommitStateSync() {
	var (
		masterState *schema.ImmutableState
		err         error
		startCh     = make(chan bool)
	)

	// Create goroutines for client waiting to query the state
	// of the followers. This is initialised before to avoid
	// spending time initialising the follower client for faster
	// state access
	var wg sync.WaitGroup
	for i := 0; i < suite.GetFollowersCount(); i++ {
		wg.Add(1)
		go func(followerID int) {
			defer wg.Done()
			ctx, client, cleanup := suite.ClientForReplica(followerID)
			defer cleanup()
			for range startCh {
				suite.Run(fmt.Sprintf("test replica sync state %d", followerID), func() {
					state, err := client.CurrentState(ctx)
					require.NoError(suite.T(), err)
					suite.Require().Equal(state.PrecommittedTxId, masterState.TxId)
					suite.Require().Equal(state.PrecommittedTxHash, masterState.TxHash)
				})
				return
			}
		}(i)
	}

	ctx, client, cleanup := suite.ClientForMaster()
	defer cleanup()

	// add multiple keys to make update the master's state quickly
	for i := 10; i < 30; i++ {
		key := fmt.Sprintf("key%d", i)
		value := fmt.Sprintf("value%d", i)
		_, err = client.Set(ctx, []byte(key), []byte(value))
		require.NoError(suite.T(), err)
	}

	// get the current precommit txn id state of master
	masterState, err = client.CurrentState(ctx)
	require.NoError(suite.T(), err)

	// close will unblock all goroutines
	close(startCh)
	wg.Wait()
}

type SyncTestMinimumFollowersSuite struct {

Review comment (suggestion):
```suggestion
			<-startCh
			suite.Run(fmt.Sprintf("test replica sync state %d", followerID), func() {
				state, err := client.CurrentState(ctx)
				require.NoError(suite.T(), err)
				suite.Require().Equal(state.PrecommittedTxId, masterState.TxId)
				suite.Require().Equal(state.PrecommittedTxHash, masterState.TxHash)
			})
```

Code after:
				suite.WaitForCommittedTx(ctx, client, tx.Id, 30*time.Second) // Longer time since replica must reestablish connection to the primary

				val, err := client.GetAt(ctx, []byte("key3"), tx.Id)
				require.NoError(suite.T(), err)
				require.Equal(suite.T(), []byte("value3"), val.Value)

				val, err = client.GetAt(ctx, []byte("key-before-restart"), txBeforeRestart.Id)
				require.NoError(suite.T(), err)
				require.Equal(suite.T(), []byte("value-before-restart"), val.Value)
			})
		}
	})
}

type SyncTestSuitePrecommitStateSync struct {
	baseReplicationTestSuite
}

func TestSyncTestSuitePrecommitStateSync(t *testing.T) {
	suite.Run(t, &SyncTestSuitePrecommitStateSync{})
}

// this function executes before the test suite begins execution
func (suite *SyncTestSuitePrecommitStateSync) SetupTest() {
	suite.baseReplicationTestSuite.SetupTest()
	suite.SetupCluster(2, 2, 0)
	suite.ValidateClusterSetup()
}

// TestPrecommitStateSync checks if the precommit state at master
// and its followers are in sync during synchronous replication
func (suite *SyncTestSuitePrecommitStateSync) TestPrecommitStateSync() {
	var (
		masterState *schema.ImmutableState
		err         error
		startCh     = make(chan bool)
	)

	ctx, client, cleanup := suite.ClientForMaster()
	defer cleanup()

	// Create goroutines for client waiting to query the state
	// of the followers. This is initialised before to avoid
	// spending time initialising the follower client for faster
	// state access
	var wg sync.WaitGroup
	for i := 0; i < suite.GetFollowersCount(); i++ {
		wg.Add(1)
		go func(followerID int) {
			defer wg.Done()
			ctx, client, cleanup := suite.ClientForReplica(followerID)