Try to commit with retries.
(&self, plan: CommitEntriesPlan)
| 290 | |
| 291 | /// Try to commit with retries. |
| 292 | async fn try_commit(&self, plan: CommitEntriesPlan) -> Result<()> { |
| 293 | let mut retry_count = 0u32; |
| 294 | let mut last_snapshot_for_dup_check: Option<Snapshot> = None; |
| 295 | let start_time_ms = current_time_millis(); |
| 296 | |
| 297 | loop { |
| 298 | let latest_snapshot = self.snapshot_manager.get_latest_snapshot().await?; |
| 299 | let resolved = self.resolve_commit(&plan, &latest_snapshot).await?; |
| 300 | |
| 301 | if resolved.entries.is_empty() { |
| 302 | break; |
| 303 | } |
| 304 | |
| 305 | // Check for duplicate commit (idempotency on retry) |
| 306 | if self |
| 307 | .is_duplicate_commit( |
| 308 | &last_snapshot_for_dup_check, |
| 309 | &latest_snapshot, |
| 310 | &resolved.kind, |
| 311 | ) |
| 312 | .await |
| 313 | { |
| 314 | break; |
| 315 | } |
| 316 | |
| 317 | let result = self.try_commit_once(resolved, &latest_snapshot).await?; |
| 318 | |
| 319 | match result { |
| 320 | true => break, |
| 321 | false => { |
| 322 | last_snapshot_for_dup_check = latest_snapshot; |
| 323 | } |
| 324 | } |
| 325 | |
| 326 | let elapsed_ms = current_time_millis() - start_time_ms; |
| 327 | if elapsed_ms > self.commit_timeout_ms || retry_count >= self.commit_max_retries { |
| 328 | let snap_id = last_snapshot_for_dup_check |
| 329 | .as_ref() |
| 330 | .map(|s| s.id() + 1) |
| 331 | .unwrap_or(1); |
| 332 | return Err(crate::Error::DataInvalid { |
| 333 | message: format!( |
| 334 | "Commit failed for snapshot {} after {} millis with {} retries, \ |
| 335 | there may exist commit conflicts between multiple jobs.", |
| 336 | snap_id, elapsed_ms, retry_count |
| 337 | ), |
| 338 | source: None, |
| 339 | }); |
| 340 | } |
| 341 | |
| 342 | self.commit_retry_wait(retry_count).await; |
| 343 | retry_count += 1; |
| 344 | } |
| 345 | |
| 346 | Ok(()) |
| 347 | } |
| 348 | |
| 349 | /// Single commit attempt. |
no test coverage detected