ValkeyCacheService: batch getIdentifiables instead of one round trip per key (streak-valkey-9)

pom.xml

@@ -16,7 +16,7 @@
 	<artifactId>objectify</artifactId>
 	<!-- Streak fork: upstream master (PR #527, Valkey cache) stamped as a fixed
 	     version for the mailfoogae Artifact Registry. Bump -streak-valkey-N for hotfixes. -->
-	<version>6.1.4-streak-valkey-8</version>
+	<version>6.1.4-streak-valkey-9</version>
 
 	<name>Objectify App Engine</name>
 	<description>The simplest convenient interface to the Google App Engine datastore</description>

src/main/java/com/googlecode/objectify/cache/valkey/ValkeyCacheService.java

@@ -292,26 +292,58 @@ public Object get(final String key) {
 	@Override
 	public Map<String, IdentifiableValue> getIdentifiables(final Collection<String> keys) {
 		final Map<String, IdentifiableValue> result = new LinkedHashMap<>();
-		for (final String key : keys) {
-			result.put(key, getIdentifiable(key));
+		if (keys.isEmpty()) {
+			return result;
 		}
-		return result;
-	}
 
-	private IdentifiableValue getIdentifiable(final String key) {
-		final byte[] bytes = rawGet(key);
-		if (bytes != null) {
-			return new ValkeyIdentifiableValue(fromCacheBytes(bytes), bytes);
+		// Per-key GETs fired concurrently (no MGET, which would CROSSSLOT on a cluster). Awaiting
+		// each GET before issuing the next would make a batch cost one serial round trip per key,
+		// which is what Objectify's entity-load path does on every batch read.
+		final Map<String, CompletableFuture<GlideString>> gets = new LinkedHashMap<>();
+		for (final String key : keys) {
+			gets.put(key, client.get(gskey(key)));
 		}
 
-		// Cold cache: bootstrap a sentinel under NX so we can later CAS against it. NX prevents
-		// us from clobbering a value another caller has just set in between our GET and our SET.
-		// TTL-bounded like every other write (defaultNxSetOptions): a read-heavy workload bootstraps
-		// a sentinel per cold key, and without expiry those persist forever on a noeviction cluster.
-		await(client.set(gskey(key), gs(NULL_VALUE), defaultNxSetOptions));
+		final Map<String, byte[]> raw = new LinkedHashMap<>();
+		gets.forEach((key, future) -> {
+			final GlideString value = await(future);
+			raw.put(key, value == null ? null : value.getBytes());
+		});
 
-		final byte[] bootstrapped = rawGet(key);
-		return bootstrapped == null ? null : new ValkeyIdentifiableValue(fromCacheBytes(bootstrapped), bootstrapped);
+		final List<String> cold = new ArrayList<>();
+		raw.forEach((key, bytes) -> {
+			if (bytes == null) {
+				cold.add(key);
+			}
+		});
+
+		if (!cold.isEmpty()) {
+			// Cold cache: bootstrap a sentinel under NX so we can later CAS against it. NX prevents
+			// us from clobbering a value another caller has just set in between our GET and our SET.
+			// TTL-bounded like every other write (defaultNxSetOptions): a read-heavy workload bootstraps
+			// a sentinel per cold key, and without expiry those persist forever on a noeviction cluster.
+			final List<CompletableFuture<String>> bootstraps = new ArrayList<>();
+			for (final String key : cold) {
+				bootstraps.add(client.set(gskey(key), gs(NULL_VALUE), defaultNxSetOptions));
+			}
+			bootstraps.forEach(ValkeyCacheService::await);
+
+			final Map<String, CompletableFuture<GlideString>> rereads = new LinkedHashMap<>();
+			for (final String key : cold) {
+				rereads.put(key, client.get(gskey(key)));
+			}
+			rereads.forEach((key, future) -> {
+				final GlideString value = await(future);
+				raw.put(key, value == null ? null : value.getBytes());
+			});
+		}
+
+		raw.forEach((key, bytes) -> {
+			if (bytes != null) {
+				result.put(key, new ValkeyIdentifiableValue(fromCacheBytes(bytes), bytes));
+			}
+		});
+		return result;
 	}
 
 	@Override

src/test/java/com/googlecode/objectify/test/valkey/ValkeyCacheServiceTests.java

@@ -131,6 +131,35 @@ void getIdentifiablesBootstrapsColdCacheWithNullSentinel() {
 		assertThat(iv.getValue()).isNull();   // sentinel decodes back to null
 	}
 
+	@Test
+	void getIdentifiablesMixesWarmAndColdKeysInOneBatch() {
+		cache.put("warm", "alpha");
+
+		final Map<String, IdentifiableValue> ivs = cache.getIdentifiables(Arrays.asList("warm", "cold", "warm2"));
+		cache.put("warm2", "gamma");   // written after the batch read; must not affect the snapshot
+
+		assertThat(ivs.keySet()).containsExactly("warm", "cold", "warm2").inOrder();
+		assertThat(ivs.get("warm").getValue()).isEqualTo("alpha");
+		assertThat(ivs.get("cold").getValue()).isNull();     // bootstrapped sentinel
+		assertThat(ivs.get("warm2").getValue()).isNull();    // bootstrapped sentinel
+
+		// Every snapshot in the batch is still a usable CAS basis; only "warm2" was stomped.
+		final Map<String, CasPut> proposed = new LinkedHashMap<>();
+		proposed.put("warm", new CasPut(ivs.get("warm"), "fresh-warm", 0));
+		proposed.put("cold", new CasPut(ivs.get("cold"), "fresh-cold", 0));
+		proposed.put("warm2", new CasPut(ivs.get("warm2"), "fresh-warm2", 0));
+
+		assertThat(cache.putIfUntouched(proposed)).containsExactly("warm", "cold");
+		assertThat(cache.get("warm")).isEqualTo("fresh-warm");
+		assertThat(cache.get("cold")).isEqualTo("fresh-cold");
+		assertThat(cache.get("warm2")).isEqualTo("gamma");
+	}
+
+	@Test
+	void getIdentifiablesOnEmptyBatchReturnsEmpty() {
+		assertThat(cache.getIdentifiables(Arrays.asList())).isEmpty();
+	}
+
 	@Test
 	void casSucceedsOnUntouchedSentinel() {
 		final IdentifiableValue iv = cache.getIdentifiables(Arrays.asList("k")).get("k");