(&mut self, permit: Option<GroupCommitPermit>)
| 285 | /// writes. |
| 286 | #[instrument(level = "debug")] |
| 287 | pub(crate) async fn try_group_commit(&mut self, permit: Option<GroupCommitPermit>) { |
| 288 | let timestamp = self.peek_local_write_ts().await; |
| 289 | let now = Timestamp::from((self.catalog().config().now)()); |
| 290 | |
| 291 | // HACK: This is a special case to allow writes to the mz_sessions table to proceed even |
| 292 | // if the timestamp oracle is ahead of the current walltime. We do this because there are |
| 293 | // some tests that mock the walltime, so it doesn't automatically advance, and updating |
| 294 | // those tests to advance the walltime while creating a connection is too much. |
| 295 | // |
| 296 | // TODO(parkmycar): Get rid of the check below when refactoring group commits. |
| 297 | let contains_internal_system_write = self |
| 298 | .pending_writes |
| 299 | .iter() |
| 300 | .any(|write| write.is_internal_system()); |
| 301 | |
| 302 | if timestamp > now && !contains_internal_system_write { |
| 303 | // Cap retry time to 1s. In cases where the system clock has retreated by |
| 304 | // some large amount of time, this prevents against then waiting for that |
| 305 | // large amount of time in case the system clock then advances back to near |
| 306 | // what it was. |
| 307 | let remaining_ms = std::cmp::min(timestamp.saturating_sub(now), 1_000.into()); |
| 308 | let internal_cmd_tx = self.internal_cmd_tx.clone(); |
| 309 | task::spawn( |
| 310 | || "group_commit_initiate", |
| 311 | async move { |
| 312 | tokio::time::sleep(Duration::from_millis(remaining_ms.into())).await; |
| 313 | // It is not an error for this task to be running after `internal_cmd_rx` is dropped. |
| 314 | let result = |
| 315 | internal_cmd_tx.send(Message::GroupCommitInitiate(Span::current(), permit)); |
| 316 | if let Err(e) = result { |
| 317 | warn!("internal_cmd_rx dropped before we could send: {:?}", e); |
| 318 | } |
| 319 | } |
| 320 | .instrument(Span::current()), |
| 321 | ); |
| 322 | } else { |
| 323 | self.group_commit(permit).await; |
| 324 | } |
| 325 | } |
| 326 | |
| 327 | /// Tries to commit all pending writes transactions at the same timestamp. |
| 328 | /// |
no test coverage detected