session: prepare: introduce fallback to all shards

Before this commit, we could unnecessarily return an error from prepare:
Let's say we have a 1-node cluster, and we have a broken connection, and
we don't retry on another - we would retry error to the user despite
possibly being able to prepare.

This commit introduces fallback logic to `Session::prepare()`:
if preparation on a single (random) connection to every node fails,
the whole preparation is retried, this time on a single connection to
every shard.

I'm a bit unhappy that this requires us to clone `statement` on the
happy path (i.e., if the on-all-nodes preparation attempt succeeds),
but I'm quite convinced it's negligible overhead.

Author: wprzytula

scylla/src/client/session.rs

@@ -1375,9 +1375,25 @@ impl Session {
             )))
         }
 
-        let on_all_nodes_result = prepare_on_all(self, statement, cluster_state, false).await?;
+        // Start by attempting preparation on a single (random) connection to every node.
+        {
+            let on_all_nodes_result =
+                prepare_on_all(self, statement.clone(), Arc::clone(&cluster_state), false).await?;
+            if let Ok(prepared) = on_all_nodes_result {
+                // We succeeded in preparing the statement on at least one node. We're done; at the same time,
+                // the background tokio task attempts preparation on remaining nodes.
+                return Ok(prepared);
+            }
+        }
 
-        on_all_nodes_result.map_err(|err| PrepareError::AllAttemptsFailed { first_attempt: err })
+        // We could have been just unlucky: we could have possibly chosen random connections all of which were defunct
+        // (one possibility is that we targeted overloaded shards).
+        // Let's try again, this time on connections to every shard. This is a "last call" fallback.
+        {
+            let on_all_shards_result = prepare_on_all(self, statement, cluster_state, true).await?;
+            on_all_shards_result
+                .map_err(|err| PrepareError::AllAttemptsFailed { first_attempt: err })
+        }
     }
 
     fn extract_partitioner_name<'a>(