ddl_vertical_postgresql: add logging to eventually!

Adding more debug logging to eventually!

Change-Id: Iaafc0fad282c3c2337fbb33705ce01f22753ed3e
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/10520
Reviewed-by: Michael Zink <[email protected]>
Tested-by: Buildkite CI

Author: altmannmarcelo

readyset-util/src/futures.rs

@@ -4,6 +4,7 @@ use std::future::Future;
 use std::panic::AssertUnwindSafe;
 use std::process;
 
+use chrono::{SecondsFormat, Utc};
 use futures::{FutureExt, TryFutureExt};
 use tracing::error;
 
@@ -153,6 +154,12 @@ where
         })
 }
 
+#[doc(hidden)]
+pub fn __eventually_warn(message: &str) {
+    let ts = Utc::now().to_rfc3339_opts(SecondsFormat::Micros, true);
+    println!("{} {}", ts, message);
+}
+
 /// Assert that the given async expression eventually succeeds after a configurable number of
 /// tries and sleeping a configurable amount between tries. Useful for testing eventually
 /// consistent parts of the system.
@@ -289,7 +296,7 @@ where
 /// ```
 #[macro_export]
 macro_rules! eventually {
-    ($(attempts: $attempts: expr,)? $(sleep: $sleep: expr,)? run_test: { $($test_body: tt)* },
+    ($(attempts: $attempts: expr,)? $(sleep: $sleep: expr,)? $(token: $token: expr,)? run_test: { $($test_body: tt)* },
             then_assert: |$test_res: pat_param| $($assert_body: tt)+) => {{
         let attempts = 40;
         let sleep = std::time::Duration::from_millis(500);
@@ -299,11 +306,21 @@ macro_rules! eventually {
 
 
         let mut res = None;
+        let token_opt: Option<&str> = None; $(let token_opt = Some($token);)?
         for attempt in 1..=(attempts - 1) {
             let $test_res = async { $($test_body)* }.await;
             match ::futures::FutureExt::catch_unwind(async { $($assert_body)* }).await {
                 Err(_) => {
-                    println!("Assertion failed on attempt {attempt}, retrying after delay...");
+                    if let Some(token) = token_opt {
+                        $crate::futures::__eventually_warn(&format!(
+                            "Assertion failed on attempt {attempt} (token: {}), retrying after delay...",
+                            token
+                        ));
+                    } else {
+                        $crate::futures::__eventually_warn(&format!(
+                            "Assertion failed on attempt {attempt}, retrying after delay..."
+                        ));
+                    }
                     tokio::time::sleep(sleep).await;
                 }
                 Ok(r) => {
@@ -323,7 +340,7 @@ macro_rules! eventually {
             }
         }
     }};
-    ($(attempts: $attempts: expr,)? $(sleep: $sleep: expr,)? run_test: { $($test_body: tt)* },
+    ($(attempts: $attempts: expr,)? $(sleep: $sleep: expr,)? $(token: $token: expr,)? run_test: { $($test_body: tt)* },
             then_assert: $($assert_body: tt)+) => {
         compile_error!(concat!("`then_assert` must be specified using closure syntax",
                                "(see `eventually` docs for detail)"))

replicators/tests/ddl_vertical_postgresql.rs

@@ -942,7 +942,7 @@ impl ModelState for DDLModelState {
 
                 let create_cache =
                     format!("CREATE CACHE ALWAYS FROM SELECT * FROM \"{table_name}\"");
-                eventually!(run_test: {
+                eventually!(token: "failed to create cache after creating table", run_test: {
                     let result = rs_conn.simple_query(&create_cache).await;
                     AssertUnwindSafe(move || result)
                 }, then_assert: |result| {
@@ -1004,7 +1004,7 @@ impl ModelState for DDLModelState {
                 rs_conn.simple_query(&query).await.unwrap();
                 pg_conn.simple_query(&query).await.unwrap();
                 let create_cache = format!("CREATE CACHE ALWAYS FROM SELECT * FROM \"{name}\"");
-                eventually!(run_test: {
+                eventually!(token: "failed to create cache after creating simple view", run_test: {
                     let result = rs_conn.simple_query(&create_cache).await;
                     AssertUnwindSafe(move || result)
                 }, then_assert: |result| {
@@ -1033,7 +1033,7 @@ impl ModelState for DDLModelState {
                 rs_conn.simple_query(&query).await.unwrap();
                 pg_conn.simple_query(&query).await.unwrap();
                 let create_cache = format!("CREATE CACHE ALWAYS FROM SELECT * FROM \"{name}\"");
-                eventually!(run_test: {
+                eventually!(token: "failed to create cache after creating join view", run_test: {
                     let result = rs_conn.simple_query(&create_cache).await;
                     AssertUnwindSafe(move || result)
                 }, then_assert: |result| {
@@ -1131,7 +1131,8 @@ impl ModelState for DDLModelState {
 
         // After each op, check that all table and view contents match
         for relation in self.tables.keys().chain(self.views.keys()) {
-            eventually!(run_test: {
+            eventually!(token: "results mismatch",
+            run_test: {
                 let rs_rows = rs_conn
                     .query(&format!("SELECT * FROM \"{relation}\""), &[])
                     .await
@@ -1154,6 +1155,10 @@ impl ModelState for DDLModelState {
                 rs_results.sort_unstable();
                 pg_results.sort_unstable();
 
+                if pg_results != rs_results {
+                    println!("results mismatch\nrs={:?}\npg={:?}", rs_results, pg_results);
+                }
+
                 assert_eq!(pg_results, rs_results);
             });
         }
@@ -1244,7 +1249,7 @@ async fn recreate_caches_using_type(
     for table in tables_using_type(tables, type_name) {
         let create_cache = format!("CREATE CACHE ALWAYS FROM SELECT * FROM \"{table}\"");
 
-        eventually!(run_test: {
+        eventually!(token: "failed to recreate cache using type", run_test: {
             let result = rs_conn.simple_query(&create_cache).await;
             AssertUnwindSafe(move || result)
         }, then_assert: |result| {