Function await_data

src/environmentd/tests/sql.rs:724–748

Asserts that the next data message is `data`. Ignores any progress messages that occur first. Returns the timestamp at which that data arrived.

(
                    client: &mut postgres::Client,
                    last_seen_ts: &mut u64,
                    data: &str,
                )

                //
                // Returns the timestamp at which that data arrived.
                fn await_data(
                    client: &mut postgres::Client,
                    last_seen_ts: &mut u64,
                    data: &str,
                ) -> u64 {
                    // We have to try several times. It might be that the FETCH gets a
                    // progress statements rather than data. We retry until we get the batch
                    // that has the data.
                    debug!("awaiting data: {data}");
                    loop {
                        let rows = client.query("FETCH 1 c1", &[]).unwrap();
                        debug!(row = ?rows.first());
                        let data_row = match rows.first() {
                            Some(row) if row.try_get::<_, String>("data").is_ok() => row,
                            _ => continue, // retry
                        };
                        assert_eq!(data_row.get::<_, bool>("mz_progressed"), false);
                        assert_eq!(data_row.get::<_, i64>("mz_diff"), 1);
                        assert_eq!(data_row.get::<_, String>("data"), data);
                        let ts = data_row.get::<_, MzTimestamp>("mz_timestamp").0;
                        assert!(ts >= *last_seen_ts);
                        *last_seen_ts = ts;
                        return ts;
                    }
                }

                // Asserts that the next message has a timestamp of at least `ts` and is a progress message.
                //