backport: set a minimum long-polling interval for Events in rpc (#8050) (#9768)

* rpc: set a minimum long-polling interval for Events (#8050)

Since the goal of reading events at the head of the event log is to satisfy a
subscription style interface, there is no point in allowing head polling with
no wait interval. The pagination case already bypasses long polling, so the
extra option is unneessary.

Set a minimum default long-polling interval for the head case.
Add a test for minimum delay.

* fix doc

Co-authored-by: M. J. Fromberger <fromberger@interchain.io>
This commit is contained in:
mmsqe
2022-12-07 18:53:39 -05:00
committed by GitHub
co-authored by M. J. Fromberger
parent 544a38da7b
commit 1b1ba41720
3 changed files with 69 additions and 5 deletions
@@ -348,8 +348,8 @@ limit.
The `wait_time` parameter is used to effect polling. If `before` is empty and
no items are available, the server will wait for up to `wait_time` for matching
items to arrive at the head of the log. If `wait_time` is zero, the server will
return whatever eligible items are available immediately.
items to arrive at the head of the log. If `wait_time` is zero or negative, the
server will wait for a default (positive) interval.
If `before` non-empty, `wait_time` is ignored: new results are only added to
the head of the log, so there is no need to wait. This allows the client to
@@ -90,6 +90,67 @@ func TestStream_lostItem(t *testing.T) {
s.stopWait()
}
func TestMinPollTime(t *testing.T) {
defer leaktest.Check(t)
s := newStreamTester(t, ``, eventlog.LogSettings{
WindowSize: 30 * time.Second,
}, nil)
s.publish("bad", "whatever")
// Waiting for an item on a log with no matching events incurs a minimum
// wait time and reports no events.
ctx := context.Background()
filter := &coretypes.EventFilter{Query: `tm.event = 'good'`}
var zero cursor.Cursor
t.Run("NoneMatch", func(t *testing.T) {
start := time.Now()
// Request a very short delay, and affirm we got the server's minimum.
rsp, err := s.Events(ctx, &coretypes.RequestEvents{
Filter: filter,
MaxItems: 1,
After: zero.String(),
Before: zero.String(),
WaitTime: 10 * time.Millisecond,
})
if err != nil {
t.Fatalf("Events failed: %v", err)
} else if elapsed := time.Since(start); elapsed < time.Second {
t.Errorf("Events returned too quickly: got %v, wanted 1s", elapsed)
} else if len(rsp.Items) != 0 {
t.Errorf("Events returned %d items, expected none", len(rsp.Items))
}
})
s.publish("good", "whatever")
// Waiting for an available matching item incurs no delay.
t.Run("SomeMatch", func(t *testing.T) {
start := time.Now()
// Request a long-ish delay and affirm we don't block for it.
// Check for this by ensuring we return sooner than the minimum delay,
// since we don't know the exact timing.
rsp, err := s.Events(ctx, &coretypes.RequestEvents{
Filter: filter,
MaxItems: 1,
After: zero.String(),
Before: zero.String(),
WaitTime: 10 * time.Second,
})
if err != nil {
t.Fatalf("Events failed: %v", err)
} else if elapsed := time.Since(start); elapsed > 500*time.Millisecond {
t.Errorf("Events returned too slowly: got %v, wanted immediate", elapsed)
} else if len(rsp.Items) == 0 {
t.Error("Events returned no items, wanted at least 1")
}
})
}
// testItem is a wrapper for comparing item results in a friendly output format
// for the cmp package.
type testItem struct {
+6 -3
View File
@@ -183,8 +183,11 @@ func EventsWithContext(ctx context.Context,
maxItems = 100
}
const minWaitTime = 1 * time.Second
const maxWaitTime = 30 * time.Second
if waitTime > maxWaitTime {
if waitTime < minWaitTime {
waitTime = minWaitTime
} else if waitTime > maxWaitTime {
waitTime = maxWaitTime
}
@@ -203,7 +206,7 @@ func EventsWithContext(ctx context.Context,
accept := func(itm *eventlog.Item) error {
// N.B. We accept up to one item more than requested, so we can tell how
// to set the "more" flag in the response.
if len(items) > maxItems {
if len(items) > maxItems || itm.Cursor.Before(after) {
return eventlog.ErrStopScan
}
match, err := query.MatchesEvents(itm.Events)
@@ -216,7 +219,7 @@ func EventsWithContext(ctx context.Context,
return nil
}
if waitTime > 0 && before.IsZero() {
if before.IsZero() {
ctx, cancel := context.WithTimeout(ctx, waitTime)
defer cancel()