From 1b1ba41720954456f395f14759f3c11c9e82a251 Mon Sep 17 00:00:00 2001 From: mmsqe Date: Thu, 8 Dec 2022 07:53:39 +0800 Subject: [PATCH] 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 --- docs/architecture/adr-075-rpc-subscription.md | 4 +- rpc/client/eventstream/eventstream_test.go | 61 +++++++++++++++++++ rpc/core/events.go | 9 ++- 3 files changed, 69 insertions(+), 5 deletions(-) diff --git a/docs/architecture/adr-075-rpc-subscription.md b/docs/architecture/adr-075-rpc-subscription.md index 28b6ab801..a838f2766 100644 --- a/docs/architecture/adr-075-rpc-subscription.md +++ b/docs/architecture/adr-075-rpc-subscription.md @@ -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 diff --git a/rpc/client/eventstream/eventstream_test.go b/rpc/client/eventstream/eventstream_test.go index 73efd104c..3ffc91893 100644 --- a/rpc/client/eventstream/eventstream_test.go +++ b/rpc/client/eventstream/eventstream_test.go @@ -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 { diff --git a/rpc/core/events.go b/rpc/core/events.go index 3d2ce477d..3ff6d4593 100644 --- a/rpc/core/events.go +++ b/rpc/core/events.go @@ -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()