From ba384e61c0951036b0c4fb394011f3498abf67ca Mon Sep 17 00:00:00 2001
From: Alex Auvolat <alex@adnab.me>
Date: Wed, 11 Jan 2023 12:03:17 +0100
Subject: [PATCH] PollRange: return immediately if no seen marker is provided

---
 doc/drafts/k2v-spec.md |  7 ++++++
 src/model/k2v/rpc.rs   | 57 +++++++++++++++++++++++++++++-------------
 2 files changed, 46 insertions(+), 18 deletions(-)

diff --git a/doc/drafts/k2v-spec.md b/doc/drafts/k2v-spec.md
index a335aee3..b3c79c08 100644
--- a/doc/drafts/k2v-spec.md
+++ b/doc/drafts/k2v-spec.md
@@ -723,6 +723,13 @@ The query body is a JSON object consisting of the following fields:
 
 The timeout can be set to any number of seconds, with a maximum of 600 seconds (10 minutes).
 
+If no seen marker is known by the caller, it can do a PollRange call
+without specifying `seenMarker`. In this case, the PollRange call will
+complete immediately, and return the current content of the range (which
+can be empty) and a seen marker to be used in further PollRange calls. This
+is the only case in which PollRange might return an HTTP 200 with an empty
+set of items.
+
 The response is either:
 
 - A HTTP 304 NOT MODIFIED response with an empty body, if the timeout expired and no changes occurred
diff --git a/src/model/k2v/rpc.rs b/src/model/k2v/rpc.rs
index 04ab3ab9..04801ebf 100644
--- a/src/model/k2v/rpc.rs
+++ b/src/model/k2v/rpc.rs
@@ -268,6 +268,8 @@ impl K2VRpcHandler {
 		seen_str: Option<String>,
 		timeout_msec: u64,
 	) -> Result<Option<(BTreeMap<String, K2VItem>, String)>, Error> {
+		let has_seen_marker = seen_str.is_some();
+
 		let mut seen = seen_str
 			.as_deref()
 			.map(RangeSeenMarker::decode)
@@ -318,7 +320,7 @@ impl K2VRpcHandler {
 			}
 		}
 
-		if new_items.is_empty() {
+		if new_items.is_empty() && has_seen_marker {
 			Ok(None)
 		} else {
 			Ok(Some((new_items, seen.encode()?)))
@@ -432,16 +434,44 @@ impl K2VRpcHandler {
 		range: &PollRange,
 		seen_str: &Option<String>,
 	) -> Result<Vec<K2VItem>, Error> {
-		let seen = seen_str
-			.as_deref()
-			.map(RangeSeenMarker::decode)
-			.transpose()?
-			.unwrap_or_default();
+		if let Some(seen_str) = seen_str {
+			let seen = RangeSeenMarker::decode(seen_str)?;
+
+			// Subscribe now to all changes on that partition,
+			// so that new items that are inserted while we are reading the range
+			// will be seen in the loop below
+			let mut chan = self.subscriptions.subscribe_partition(&range.partition);
+
+			// Check for the presence of any new items already stored in the item table
+			let mut new_items = self.poll_range_read_range(range, &seen)?;
+
+			// If we found no new items, wait for a matching item to arrive
+			// on the channel
+			while new_items.is_empty() {
+				let item = chan.recv().await?;
+				if range.matches(&item) && seen.is_new_item(&item) {
+					new_items.push(item);
+				}
+			}
+
+			Ok(new_items)
+		} else {
+			// If no seen marker was specified, we do not poll for anything.
+			// We return immediately with the set of known items (even if
+			// it is empty), which will give the client an inital view of
+			// the dataset and an initial seen marker for further
+			// PollRange calls.
+			self.poll_range_read_range(range, &RangeSeenMarker::default())
+		}
+	}
+
+	fn poll_range_read_range(
+		&self,
+		range: &PollRange,
+		seen: &RangeSeenMarker,
+	) -> Result<Vec<K2VItem>, Error> {
 		let mut new_items = vec![];
 
-		let mut chan = self.subscriptions.subscribe_partition(&range.partition);
-
-		// Read current state of the specified range to check new items
 		let partition_hash = range.partition.hash();
 		let first_key = match &range.start {
 			None => partition_hash.to_vec(),
@@ -461,15 +491,6 @@ impl K2VRpcHandler {
 			}
 		}
 
-		// If we found no new items, wait for a matching item to arrive
-		// on the channel
-		while new_items.is_empty() {
-			let item = chan.recv().await?;
-			if range.matches(&item) && seen.is_new_item(&item) {
-				new_items.push(item);
-			}
-		}
-
 		Ok(new_items)
 	}
 }