fix(storage): cancel paginated value reads

Makefile

@@ -282,7 +282,15 @@ build-bench-cache-service-delete: proto
 build-bench-cache-service-list-with-values: proto
 	@test -n "$(PERFLOOP_BUILD_OUTPUT_DIR)" || { echo "PERFLOOP_BUILD_OUTPUT_DIR is required"; exit 1; }
 	@mkdir -p "$(PERFLOOP_BUILD_OUTPUT_DIR)"
-	@cd server && CGO_ENABLED=1 CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" go test $(LDFLAGS) -c -o "$(PERFLOOP_BUILD_OUTPUT_DIR)/cache-service-list-with-values.test" ./service
+	@cd server && CGO_ENABLED=1 CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" go test $(LDFLAGS) -tags=ocache_benchmark -c -o "$(PERFLOOP_BUILD_OUTPUT_DIR)/cache-service-list-with-values.test" ./service
+
+.PHONY: test-bench-cache-service-list-with-values-cancellation
+test-bench-cache-service-list-with-values-cancellation: proto
+	@cd server && CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" go test $(LDFLAGS) -tags=ocache_benchmark -v -timeout 60s -run '^TestCacheServiceListWithValuesCancellationReleasesBlockedRead$$' ./service
+
+.PHONY: test-race-bench-cache-service-list-with-values-cancellation
+test-race-bench-cache-service-list-with-values-cancellation: proto
+	@cd server && CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" go test $(LDFLAGS) -race -tags=ocache_benchmark -v -timeout 60s -run '^TestCacheServiceListWithValuesCancellationReleasesBlockedRead$$' ./service
 
 .PHONY: run-background
 run-background:

coordinator/benchmark.go

@@ -0,0 +1,29 @@
+// Copyright 2026 Tigris Data, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+//go:build ocache_benchmark
+
+package coordinator
+
+import (
+	"github.com/tigrisdata/ocache/coordinator/ring"
+)
+
+// NewBenchmarkCoordinator wires a running in-memory ring to the normal routing
+// path without starting memberlist. It is used only by the cluster list
+// cancellation benchmark.
+func NewBenchmarkCoordinator(ringManager *ring.RingManager, localID string) *Coordinator {
+	return &Coordinator{
+		config: &Config{
+			Enabled:  true,
+			MyNodeID: localID,
+			LifecyclerConfig: ring.LifecyclerConfig{
+				RingConfig: ring.Config{ReplicationFactor: 1},
+			},
+		},
+		ringManager: ringManager,
+		router:      NewRouter(ringManager, localID),
+		stopCh:      make(chan struct{}),
+		errCh:       make(chan error, 1),
+	}
+}

coordinator/ring/topology_benchmark.go

@@ -1,7 +1,7 @@
 // Copyright 2026 Tigris Data, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-//go:build ocache_topology_benchmark
+//go:build ocache_benchmark || ocache_topology_benchmark
 
 package ring
 

server/operations/list_with_values.go

@@ -60,18 +60,29 @@ func (o *Operations) ListLocalWithValues(ctx context.Context, prefix string, lim
 	var lastKey string
 	var hasMore bool
 
+	if err := ctx.Err(); err != nil {
+		return nil, "", false, err
+	}
+
 	err := retry.Do(ctx, retry.DefaultConfig(), "ListKeyValuesWithPagination", func() error {
 		var listErr error
-		storageEntries, lastKey, hasMore, listErr = o.storage.ListKeyValuesWithPagination(prefix, startKey, limit)
+		storageEntries, lastKey, hasMore, listErr = o.storage.ListKeyValuesWithPagination(ctx, prefix, startKey, limit)
 		return listErr
 	})
 	if err != nil {
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			return nil, "", false, ctxErr
+		}
 		return nil, "", false, err
 	}
 
-	// Convert storage entries to proto entries
+	// Convert storage entries to proto entries. A canceled request never
+	// publishes a partially converted page.
 	entries := make([]*pb.KeyValue, len(storageEntries))
 	for i, e := range storageEntries {
+		if err := ctx.Err(); err != nil {
+			return nil, "", false, err
+		}
 		entries[i] = &pb.KeyValue{
 			Key:          e.Key,
 			Value:        e.Value,

server/service/list_with_values.go

@@ -12,11 +12,15 @@ import (
 	zlog "github.com/rs/zerolog/log"
 	"github.com/tigrisdata/ocache/common/metrics"
 	pb "github.com/tigrisdata/ocache/proto"
+	"github.com/tigrisdata/ocache/storage/benchio"
 )
 
 // ListWithValues implements the public ListWithValues RPC.
 // Returns sorted, paginated key-value pairs with continuation token.
 func (s *CacheService) ListWithValues(ctx context.Context, req *pb.ListRequest) (*pb.ListWithValuesResponse, error) {
+	endBenchmarkHandler := benchio.BeginListHandlerForBenchmark()
+	defer endBenchmarkHandler()
+
 	start := time.Now()
 	defer func() {
 		metrics.RPCDuration.WithLabelValues("ListWithValues").Observe(float64(time.Since(start).Milliseconds()))
@@ -52,6 +56,9 @@ func (s *CacheService) ListWithValues(ctx context.Context, req *pb.ListRequest)
 // ListLocalWithValues implements the internal node-local ListWithValues RPC.
 // Returns sorted, paginated key-value pairs from this node's local storage.
 func (s *CacheService) ListLocalWithValues(ctx context.Context, req *pb.ListRequest) (*pb.ListWithValuesResponse, error) {
+	endBenchmarkHandler := benchio.BeginListHandlerForBenchmark()
+	defer endBenchmarkHandler()
+
 	start := time.Now()
 	defer func() {
 		metrics.RPCDuration.WithLabelValues("ListLocalWithValues").Observe(float64(time.Since(start).Milliseconds()))