unified storage: check for iterator errors after each call to iter.Next. (#102804)

* unified storage: check for iterator errors after each call to iter.Next.
* Extracted test to separate method, add 500 events before listing.

pkg/storage/unified/resource/search.go

@@ -563,7 +563,7 @@ func (s *searchSupport) build(ctx context.Context, nsr NamespacedResource, size
 					return err
 				}
 			}
-			return err
+			return iter.Error()
 		})
 		return rv, err
 	})

pkg/storage/unified/resource/server.go

@@ -18,6 +18,7 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 
 	claims "github.com/grafana/authlib/types"
+
 	"github.com/grafana/grafana/pkg/apimachinery/utils"
 )
 
@@ -32,15 +33,18 @@ type ResourceServer interface {
 }
 
 type ListIterator interface {
+	// Next advances iterator and returns true if there is next value is available from the iterator.
+	// Error() should be checked after every call of Next(), even when Next() returns true.
 	Next() bool // sql.Rows
 
-	// Iterator error (if exts)
+	// Error returns iterator error, if any. This should be checked after any Next() call.
+	// (Some iterator implementations return true from Next, but also set the error at the same time).
 	Error() error
 
-	// The token that can be used to start iterating *after* this item
+	// ContinueToken returns the token that can be used to start iterating *after* this item
 	ContinueToken() string
 
-	// The token that can be used to start iterating *before* this item
+	// ContinueTokenWithCurrentRV returns the token that can be used to start iterating *before* this item
 	ContinueTokenWithCurrentRV() string
 
 	// ResourceVersion of the current item
@@ -763,11 +767,10 @@ func (s *server) List(ctx context.Context, req *ListRequest) (*ListResponse, err
 				if iter.Next() {
 					rsp.NextPageToken = t
 				}
-
+				return iter.Error()
-				break
 			}
 		}
-		return nil
+		return iter.Error()
 	})
 	if err != nil {
 		rsp.Error = AsErrorResult(err)
@@ -871,7 +874,7 @@ func (s *server) Watch(req *WatchRequest, srv ResourceStore_WatchServer) error {
 					return err
 				}
 			}
-			return nil
+			return iter.Error()
 		})
 		if err != nil {
 			return err

pkg/storage/unified/testing/storage_backend.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/grafana/authlib/authn"
 	"github.com/grafana/authlib/types"
+
 	"github.com/grafana/grafana/pkg/apimachinery/utils"
 	"github.com/grafana/grafana/pkg/storage/unified/resource"
 	"github.com/grafana/grafana/pkg/util/testutil"
@@ -26,13 +27,14 @@ import (
 
 // Test names for the storage backend test suite
 const (
-	TestHappyPath         = "happy path"
+	TestHappyPath                 = "happy path"
-	TestWatchWriteEvents  = "watch write events from latest"
+	TestWatchWriteEvents          = "watch write events from latest"
-	TestList              = "list"
+	TestList                      = "list"
-	TestBlobSupport       = "blob support"
+	TestBlobSupport               = "blob support"
-	TestGetResourceStats  = "get resource stats"
+	TestGetResourceStats          = "get resource stats"
-	TestListHistory       = "list history"
+	TestListHistory               = "list history"
-	TestCreateNewResource = "create new resource"
+	TestListHistoryErrorReporting = "list history error reporting"
+	TestCreateNewResource         = "create new resource"
 )
 
 type NewBackendFunc func(ctx context.Context) resource.StorageBackend
@@ -75,6 +77,7 @@ func RunStorageBackendTest(t *testing.T, newBackend NewBackendFunc, opts *TestOp
 		{TestBlobSupport, runTestIntegrationBlobSupport},
 		{TestGetResourceStats, runTestIntegrationBackendGetResourceStats},
 		{TestListHistory, runTestIntegrationBackendListHistory},
+		{TestListHistoryErrorReporting, runTestIntegrationBackendListHistoryErrorReporting},
 		{TestCreateNewResource, runTestIntegrationBackendCreateNewResource},
 	}
 
@@ -476,7 +479,7 @@ func runTestIntegrationBackendList(t *testing.T, backend resource.StorageBackend
 }
 
 func runTestIntegrationBackendListHistory(t *testing.T, backend resource.StorageBackend, nsPrefix string) {
-	ctx := testutil.NewTestContext(t, time.Now().Add(5*time.Second))
+	ctx := testutil.NewTestContext(t, time.Now().Add(30*time.Second))
 	server := newServer(t, backend)
 	ns := nsPrefix + "-ns1"
 	rv1, _ := writeEvent(ctx, backend, "item1", resource.WatchEvent_ADDED, WithNamespace(ns))
@@ -839,6 +842,58 @@ func runTestIntegrationBackendListHistory(t *testing.T, backend resource.Storage
 	})
 }
 
+func runTestIntegrationBackendListHistoryErrorReporting(t *testing.T, backend resource.StorageBackend, nsPrefix string) {
+	ctx := testutil.NewTestContext(t, time.Now().Add(30*time.Second))
+	server := newServer(t, backend)
+
+	ns := nsPrefix + "-short"
+	const (
+		name         = "it1"
+		group        = "group"
+		resourceName = "resource"
+	)
+
+	start := time.Now()
+	origRv, _ := writeEvent(ctx, backend, name, resource.WatchEvent_ADDED, WithNamespace(ns), WithGroup(group), WithResource(resourceName))
+	require.Greater(t, origRv, int64(0))
+
+	const events = 500
+	prevRv := origRv
+	for i := 0; i < events; i++ {
+		rv, err := writeEvent(ctx, backend, name, resource.WatchEvent_MODIFIED, WithNamespace(ns), WithGroup(group), WithResource(resourceName))
+		require.NoError(t, err)
+		require.Greater(t, rv, prevRv)
+		prevRv = rv
+	}
+	t.Log("added events in ", time.Since(start))
+
+	req := &resource.ListRequest{
+		Limit:           2 * events,
+		Source:          resource.ListRequest_HISTORY,
+		ResourceVersion: origRv,
+		VersionMatchV2:  resource.ResourceVersionMatchV2_NotOlderThan,
+		Options: &resource.ListOptions{
+			Key: &resource.ResourceKey{
+				Namespace: ns,
+				Group:     group,
+				Resource:  resourceName,
+				Name:      name,
+			},
+		},
+	}
+
+	shortContext, cancel := context.WithTimeout(ctx, 1*time.Millisecond)
+	defer cancel()
+
+	res, err := server.List(shortContext, req)
+	// We expect context deadline error, but it may be reported as a res.Error object.
+	t.Log("list error:", err)
+	if res != nil {
+		t.Log("iterator error:", res.Error)
+	}
+	require.True(t, err != nil || (res != nil && res.Error != nil))
+}
+
 func runTestIntegrationBlobSupport(t *testing.T, backend resource.StorageBackend, nsPrefix string) {
 	ctx := testutil.NewTestContext(t, time.Now().Add(5*time.Second))
 	server := newServer(t, backend)