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

Code before:
			})
		}
	})
}

// TestPrecommitStateSync checks if the precommit state at master
// and its followers are in sync during synchronous replication
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)

Review comment (suggestion):
This would not run at all because I don't think range checks for nil

https://go.dev/play/p/pJ1sU5I9x6n

Code after:
			suite.Run(fmt.Sprintf("check follower %d", i), func() {
				ctx, client, cleanup := suite.ClientForReplica(i)
				defer cleanup()

				// Tests are flaky because it takes time to commit the
				// precommitted TX, so this function just ensures the state
				// is in sync between master and follower
				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