cmd/tools/segment-verify: fixes and more tests

* Disallow too large listing limit, which would cause a lot of memory to
  be consumed.
* Fix throttling logic and add a test.
* Fix read error handling; depending on the concurrency it can return
  the NotFound status either in the Read or Close.

Change-Id: I778f04a5961988b2480df5c7faaa22393fc5d760

cmd/tools/segment-verify/verify.go

@@ -67,13 +67,14 @@ func (service *NodeVerifier) Verify(ctx context.Context, target storj.NodeURL, s
 	}
 	defer func() { _ = client.Close() }()
 
-	for _, segment := range segments {
+	for i, segment := range segments {
 		downloadStart := time.Now()
 		err := service.verifySegment(ctx, client, target, segment)
 		if err != nil {
 			return Error.Wrap(err)
 		}
-		if throttle := time.Since(downloadStart); !ignoreThrottle && throttle > 0 {
+		throttle := service.config.RequestThrottle - time.Since(downloadStart)
+		if !ignoreThrottle && throttle > 0 && i < len(segments)-1 {
 			if !sync2.Sleep(ctx, throttle) {
 				return Error.Wrap(ctx.Err())
 			}
@@ -122,20 +123,12 @@ func (service *NodeVerifier) verifySegment(ctx context.Context, client *piecesto
 			zap.Error(err))
 		return ErrNodeOffline.Wrap(err)
 	}
-	defer func() {
-		errClose := downloader.Close()
-		if errClose != nil {
-			// TODO: should we try reconnect in this case?
-			service.log.Error("close failed",
-				zap.String("stream-id", segment.StreamID.String()),
-				zap.Uint64("position", segment.Position.Encode()),
-				zap.Error(err))
-			err = errs.Combine(err, ErrNodeOffline.Wrap(errClose))
-		}
-	}()
 
 	buf := [1]byte{}
-	_, err = io.ReadFull(downloader, buf[:])
+	_, errRead := io.ReadFull(downloader, buf[:])
+	errClose := downloader.Close()
+
+	err = errs.Combine(errClose, errRead)
 	if err != nil {
 		if errs2.IsRPC(err, rpcstatus.NotFound) {
 			service.log.Info("segment not found",
@@ -146,7 +139,7 @@ func (service *NodeVerifier) verifySegment(ctx context.Context, client *piecesto
 			return nil
 		}
 
-		service.log.Error("read failed",
+		service.log.Error("read/close failed",
 			zap.String("stream-id", segment.StreamID.String()),
 			zap.Uint64("position", segment.Position.Encode()),
 			zap.Error(err))

cmd/tools/segment-verify/verify_test.go

@@ -25,14 +25,16 @@ func TestVerifier(t *testing.T) {
 	}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
 		satellite := planet.Satellites[0]
 
+		config := segmentverify.VerifierConfig{
+			PerPieceTimeout:    time.Second,
+			OrderRetryThrottle: 500 * time.Millisecond,
+			RequestThrottle:    500 * time.Millisecond,
+		}
 		service := segmentverify.NewVerifier(
 			planet.Log().Named("verifier"),
 			satellite.Dialer,
 			satellite.Orders.Service,
-			segmentverify.VerifierConfig{
-				PerPieceTimeout:    800 * time.Millisecond,
-				OrderRetryThrottle: 50 * time.Millisecond,
-			})
+			config)
 
 		// upload some data
 		data := testrand.Bytes(8 * memory.KiB)
@@ -46,7 +48,7 @@ func TestVerifier(t *testing.T) {
 		result, err := satellite.Metabase.DB.ListVerifySegments(ctx, metabase.ListVerifySegments{
 			CursorStreamID: uuid.UUID{},
 			CursorPosition: metabase.SegmentPosition{},
-			Limit:          100000000,
+			Limit:          10000,
 		})
 		require.NoError(t, err)
 
@@ -66,6 +68,10 @@ func TestVerifier(t *testing.T) {
 		}
 
 		// segment not found
+		validSegment0 := &segmentverify.Segment{
+			VerifySegment: result.Segments[0],
+			Status:        segmentverify.Status{Retry: 1},
+		}
 		missingSegment := &segmentverify.Segment{
 			VerifySegment: metabase.VerifySegment{
 				StreamID:    testrand.UUID(),
@@ -75,10 +81,25 @@ func TestVerifier(t *testing.T) {
 			},
 			Status: segmentverify.Status{Retry: 1},
 		}
+		validSegment1 := &segmentverify.Segment{
+			VerifySegment: result.Segments[1],
+			Status:        segmentverify.Status{Retry: 1},
+		}
 
-		err = service.Verify(ctx, planet.StorageNodes[0].NodeURL(), []*segmentverify.Segment{missingSegment}, true)
+		err = service.Verify(ctx, planet.StorageNodes[0].NodeURL(),
+			[]*segmentverify.Segment{validSegment0, missingSegment, validSegment1}, true)
 		require.NoError(t, err)
-		require.Equal(t, segmentverify.Status{Found: 0, NotFound: 1, Retry: 0}, missingSegment.Status)
+		require.Equal(t, segmentverify.Status{Found: 1}, validSegment0.Status)
+		require.Equal(t, segmentverify.Status{NotFound: 1}, missingSegment.Status)
+		require.Equal(t, segmentverify.Status{Found: 1}, validSegment1.Status)
+
+		// Test throttling
+		verifyStart := time.Now()
+		const throttleN = 5
+		err = service.Verify(ctx, planet.StorageNodes[0].NodeURL(), validSegments[:throttleN], false)
+		require.NoError(t, err)
+		verifyDuration := time.Since(verifyStart)
+		require.Greater(t, verifyDuration, config.RequestThrottle*(throttleN-1))
 
 		// TODO: test download timeout
 

satellite/metabase/list_verify_segments.go

@@ -12,6 +12,9 @@ import (
 	"storj.io/private/tagsql"
 )
 
+// ListVerifyLimit is the maximum number of items the client can request for listing.
+const ListVerifyLimit = intLimitRange(100000)
+
 // ListVerifySegments contains arguments necessary for listing stream segments.
 type ListVerifySegments struct {
 	CursorStreamID uuid.UUID
@@ -48,6 +51,7 @@ func (db *DB) ListVerifySegments(ctx context.Context, opts ListVerifySegments) (
 	if opts.Limit <= 0 {
 		return ListVerifySegmentsResult{}, Error.New("Invalid limit: %d", opts.Limit)
 	}
+	ListVerifyLimit.Ensure(&opts.Limit)
 	result.Segments = make([]VerifySegment, 0, opts.Limit)
 
 	err = withRows(db.db.QueryContext(ctx, `