Reject watch request with -1 revision and make rangeEvents safe against negative revision

Signed-off-by: Marek Siarkowicz <[email protected]>

Author: serathius

server/etcdserver/api/v3rpc/watch.go

@@ -272,6 +272,22 @@ func (sws *serverWatchStream) recvLoop() error {
 				// support  >= key queries
 				creq.RangeEnd = []byte{}
 			}
+			if creq.StartRevision < 0 {
+				wr := &pb.WatchResponse{
+					Header:       sws.newResponseHeader(sws.watchStream.Rev()),
+					WatchId:      clientv3.InvalidWatchID,
+					Canceled:     true,
+					Created:      true,
+					CancelReason: rpctypes.ErrCompacted.Error(),
+				}
+
+				select {
+				case sws.ctrlStream <- wr:
+					continue
+				case <-sws.closec:
+					return nil
+				}
+			}
 
 			err := sws.isWatchPermitted(creq)
 			if err != nil {

server/proxy/grpcproxy/watch.go

@@ -235,6 +235,17 @@ func (wps *watchProxyStream) recvLoop() error {
 		case *pb.WatchRequest_CreateRequest:
 			cr := uv.CreateRequest
 
+			if cr.StartRevision < 0 {
+				wps.watchCh <- &pb.WatchResponse{
+					Header:       &pb.ResponseHeader{},
+					WatchId:      clientv3.InvalidWatchID,
+					Created:      true,
+					Canceled:     true,
+					CancelReason: rpctypes.ErrCompacted.Error(),
+				}
+				continue
+			}
+
 			if err := wps.checkPermissionForWatch(cr.Key, cr.RangeEnd); err != nil {
 				wps.watchCh <- &pb.WatchResponse{
 					Header:       &pb.ResponseHeader{},

server/storage/mvcc/watchable_store.go

@@ -447,6 +447,10 @@ func rangeEventsWithReuse(lg *zap.Logger, b backend.Backend, evs []mvccpb.Event,
 
 // rangeEvents returns events in range [minRev, maxRev).
 func rangeEvents(lg *zap.Logger, b backend.Backend, minRev, maxRev int64) []mvccpb.Event {
+	if minRev < 0 {
+		lg.Warn("Unexpected negative revision range start", zap.Int64("minRev", minRev))
+		minRev = 0
+	}
 	minBytes, maxBytes := NewRevBytes(), NewRevBytes()
 	minBytes = RevToBytes(Revision{Main: minRev}, minBytes)
 	maxBytes = RevToBytes(Revision{Main: maxRev}, maxBytes)

server/storage/mvcc/watchable_store_test.go

@@ -405,13 +405,14 @@ func TestRangeEvents(t *testing.T) {
 		expectEvents []mvccpb.Event
 	}{
 		// maxRev, top to bottom
-		{minRev: 2, maxRev: 6, expectEvents: expectEvents[0:5]},
-		{minRev: 2, maxRev: 5, expectEvents: expectEvents[0:3]},
-		{minRev: 2, maxRev: 4, expectEvents: expectEvents[0:2]},
-		{minRev: 2, maxRev: 3, expectEvents: expectEvents[0:1]},
-		{minRev: 2, maxRev: 2, expectEvents: expectEvents[0:0]},
+		{minRev: -1, maxRev: 6, expectEvents: expectEvents[0:5]},
+		{minRev: -1, maxRev: 5, expectEvents: expectEvents[0:3]},
+		{minRev: -1, maxRev: 4, expectEvents: expectEvents[0:2]},
+		{minRev: -1, maxRev: 3, expectEvents: expectEvents[0:1]},
+		{minRev: -1, maxRev: 2, expectEvents: expectEvents[0:0]},
 
 		// minRev, bottom to top
+		{minRev: -1, maxRev: 6, expectEvents: expectEvents[0:5]},
 		{minRev: 2, maxRev: 6, expectEvents: expectEvents[0:5]},
 		{minRev: 3, maxRev: 6, expectEvents: expectEvents[1:5]},
 		{minRev: 4, maxRev: 6, expectEvents: expectEvents[2:5]},

tests/integration/clientv3/watch_test.go

@@ -25,6 +25,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/metadata"
 
@@ -1133,3 +1134,80 @@ func testWatchClose(t *testing.T, wctx *watchctx) {
 	wresp, ok := <-wch
 	require.Falsef(t, ok, "read wch got %v; expected closed channel", wresp)
 }
+
+func TestWatch(t *testing.T) {
+	integration2.BeforeTest(t)
+	clus := integration2.NewCluster(t, &integration2.ClusterConfig{Size: 1})
+	t.Cleanup(func() { clus.Terminate(t) })
+	client := clus.Client(0)
+
+	tcs := []struct {
+		name       string
+		key        string
+		opts       []clientv3.OpOption
+		wantError  error
+		wantEvents []*clientv3.Event
+	}{
+		{
+			name:      "Watch with negative revision",
+			key:       "/",
+			opts:      []clientv3.OpOption{clientv3.WithRev(-1)},
+			wantError: rpctypes.ErrCompacted,
+		},
+		{
+			name: "Watch with zero revision",
+			key:  "/",
+			opts: []clientv3.OpOption{clientv3.WithRev(0)},
+		},
+		{
+			name: "Watch with positive revision",
+			key:  "/",
+			opts: []clientv3.OpOption{clientv3.WithRev(1)},
+		},
+	}
+	ctx := t.Context()
+
+	t.Log("Open watches")
+	watches := make([]clientv3.WatchChan, len(tcs))
+	for i, tc := range tcs {
+		watchCtx, cancel := context.WithTimeout(ctx, time.Second)
+		defer cancel()
+		watches[i] = client.Watch(watchCtx, tc.key, tc.opts...)
+	}
+
+	t.Log("Validate")
+	for i, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			events, err := collectEvents(ctx, watches[i])
+			if tc.wantError == nil {
+				require.NoError(t, err)
+			} else {
+				require.ErrorContains(t, err, tc.wantError.Error())
+			}
+			if diff := cmp.Diff(tc.wantEvents, events); diff != "" {
+				t.Errorf("unexpected events (-want +got):\n%s", diff)
+			}
+		})
+	}
+}
+
+func collectEvents(ctx context.Context, watch clientv3.WatchChan) (events []*clientv3.Event, err error) {
+	for {
+		select {
+		case resp, ok := <-watch:
+			if !ok {
+				return events, nil
+			}
+			err := resp.Err()
+			if err != nil {
+				return events, err
+			}
+			events = append(events, resp.Events...)
+		case <-ctx.Done():
+			return events, ctx.Err()
+		// Watch resync interval * 1.5
+		case <-time.After(150 * time.Millisecond):
+			return events, nil
+		}
+	}
+}