|
|
|
@ -233,13 +233,10 @@ func TestRPC_blockingQuery(t *testing.T) {
|
|
|
|
|
defer os.RemoveAll(dir)
|
|
|
|
|
defer s.Shutdown()
|
|
|
|
|
|
|
|
|
|
require := require.New(t)
|
|
|
|
|
assert := assert.New(t)
|
|
|
|
|
|
|
|
|
|
// Perform a non-blocking query. Note that it's significant that the meta has
|
|
|
|
|
// a zero index in response - the implied opts.MinQueryIndex is also zero but
|
|
|
|
|
// this should not block still.
|
|
|
|
|
{
|
|
|
|
|
t.Run("non-blocking query", func(t *testing.T) {
|
|
|
|
|
var opts structs.QueryOptions
|
|
|
|
|
var meta structs.QueryMeta
|
|
|
|
|
var calls int
|
|
|
|
@ -247,16 +244,13 @@ func TestRPC_blockingQuery(t *testing.T) {
|
|
|
|
|
calls++
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
if err := s.blockingQuery(&opts, &meta, fn); err != nil {
|
|
|
|
|
t.Fatalf("err: %v", err)
|
|
|
|
|
}
|
|
|
|
|
if calls != 1 {
|
|
|
|
|
t.Fatalf("bad: %d", calls)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
|
|
|
require.NoError(t, err)
|
|
|
|
|
require.Equal(t, 1, calls)
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
// Perform a blocking query that gets woken up and loops around once.
|
|
|
|
|
{
|
|
|
|
|
t.Run("blocking query - single loop", func(t *testing.T) {
|
|
|
|
|
opts := structs.QueryOptions{
|
|
|
|
|
MinQueryIndex: 3,
|
|
|
|
|
}
|
|
|
|
@ -275,13 +269,10 @@ func TestRPC_blockingQuery(t *testing.T) {
|
|
|
|
|
calls++
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
if err := s.blockingQuery(&opts, &meta, fn); err != nil {
|
|
|
|
|
t.Fatalf("err: %v", err)
|
|
|
|
|
}
|
|
|
|
|
if calls != 2 {
|
|
|
|
|
t.Fatalf("bad: %d", calls)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
|
|
|
require.NoError(t, err)
|
|
|
|
|
require.Equal(t, 2, calls)
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
// Perform a blocking query that returns a zero index from blocking func (e.g.
|
|
|
|
|
// no state yet). This should still return an empty response immediately, but
|
|
|
|
@ -292,7 +283,7 @@ func TestRPC_blockingQuery(t *testing.T) {
|
|
|
|
|
// covered by tests but eventually when hit in the wild causes blocking
|
|
|
|
|
// clients to busy loop and burn CPU. This test ensure that blockingQuery
|
|
|
|
|
// systematically does the right thing to prevent future bugs like that.
|
|
|
|
|
{
|
|
|
|
|
t.Run("blocking query with 0 modifyIndex from state func", func(t *testing.T) {
|
|
|
|
|
opts := structs.QueryOptions{
|
|
|
|
|
MinQueryIndex: 0,
|
|
|
|
|
}
|
|
|
|
@ -311,9 +302,9 @@ func TestRPC_blockingQuery(t *testing.T) {
|
|
|
|
|
calls++
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
require.NoError(s.blockingQuery(&opts, &meta, fn))
|
|
|
|
|
assert.Equal(1, calls)
|
|
|
|
|
assert.Equal(uint64(1), meta.Index,
|
|
|
|
|
require.NoError(t, s.blockingQuery(&opts, &meta, fn))
|
|
|
|
|
assert.Equal(t, 1, calls)
|
|
|
|
|
assert.Equal(t, uint64(1), meta.Index,
|
|
|
|
|
"expect fake index of 1 to force client to block on next update")
|
|
|
|
|
|
|
|
|
|
// Simulate client making next request
|
|
|
|
@ -322,19 +313,19 @@ func TestRPC_blockingQuery(t *testing.T) {
|
|
|
|
|
|
|
|
|
|
// This time we should block even though the func returns index 0 still
|
|
|
|
|
t0 := time.Now()
|
|
|
|
|
require.NoError(s.blockingQuery(&opts, &meta, fn))
|
|
|
|
|
require.NoError(t, s.blockingQuery(&opts, &meta, fn))
|
|
|
|
|
t1 := time.Now()
|
|
|
|
|
assert.Equal(2, calls)
|
|
|
|
|
assert.Equal(uint64(1), meta.Index,
|
|
|
|
|
assert.Equal(t, 2, calls)
|
|
|
|
|
assert.Equal(t, uint64(1), meta.Index,
|
|
|
|
|
"expect fake index of 1 to force client to block on next update")
|
|
|
|
|
assert.True(t1.Sub(t0) > 20*time.Millisecond,
|
|
|
|
|
assert.True(t, t1.Sub(t0) > 20*time.Millisecond,
|
|
|
|
|
"should have actually blocked waiting for timeout")
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
// Perform a query that blocks and gets interrupted when the state store
|
|
|
|
|
// is abandoned.
|
|
|
|
|
{
|
|
|
|
|
t.Run("blocking query interrupted by abandonCh", func(t *testing.T) {
|
|
|
|
|
opts := structs.QueryOptions{
|
|
|
|
|
MinQueryIndex: 3,
|
|
|
|
|
}
|
|
|
|
@ -363,13 +354,10 @@ func TestRPC_blockingQuery(t *testing.T) {
|
|
|
|
|
calls++
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
if err := s.blockingQuery(&opts, &meta, fn); err != nil {
|
|
|
|
|
t.Fatalf("err: %v", err)
|
|
|
|
|
}
|
|
|
|
|
if calls != 1 {
|
|
|
|
|
t.Fatalf("bad: %d", calls)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
|
|
|
require.NoError(t, err)
|
|
|
|
|
require.Equal(t, 1, calls)
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
t.Run("ResultsFilteredByACLs is reset for unauthenticated calls", func(t *testing.T) {
|
|
|
|
|
opts := structs.QueryOptions{
|
|
|
|
@ -382,13 +370,13 @@ func TestRPC_blockingQuery(t *testing.T) {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
err := s.blockingQuery(&opts, &meta, fn)
|
|
|
|
|
require.NoError(err)
|
|
|
|
|
require.False(meta.ResultsFilteredByACLs, "ResultsFilteredByACLs should be reset for unauthenticated calls")
|
|
|
|
|
require.NoError(t, err)
|
|
|
|
|
require.False(t, meta.ResultsFilteredByACLs, "ResultsFilteredByACLs should be reset for unauthenticated calls")
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
t.Run("ResultsFilteredByACLs is honored for authenticated calls", func(t *testing.T) {
|
|
|
|
|
token, err := lib.GenerateUUID(nil)
|
|
|
|
|
require.NoError(err)
|
|
|
|
|
require.NoError(t, err)
|
|
|
|
|
|
|
|
|
|
opts := structs.QueryOptions{
|
|
|
|
|
Token: token,
|
|
|
|
@ -400,8 +388,8 @@ func TestRPC_blockingQuery(t *testing.T) {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
err = s.blockingQuery(&opts, &meta, fn)
|
|
|
|
|
require.NoError(err)
|
|
|
|
|
require.True(meta.ResultsFilteredByACLs, "ResultsFilteredByACLs should be honored for authenticated calls")
|
|
|
|
|
require.NoError(t, err)
|
|
|
|
|
require.True(t, meta.ResultsFilteredByACLs, "ResultsFilteredByACLs should be honored for authenticated calls")
|
|
|
|
|
})
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|