feat: retry failed transaction commit

src/iceberg/test/CMakeLists.txt

@@ -119,6 +119,7 @@ add_iceberg_test(util_test
                  location_util_test.cc
                  roaring_position_bitmap_test.cc
                  position_delete_index_test.cc
+                 retry_util_test.cc
                  string_util_test.cc
                  struct_like_set_test.cc
                  transform_util_test.cc

src/iceberg/test/meson.build

@@ -91,6 +91,7 @@ iceberg_tests = {
             'formatter_test.cc',
             'location_util_test.cc',
             'position_delete_index_test.cc',
+            'retry_util_test.cc',
             'roaring_position_bitmap_test.cc',
             'string_util_test.cc',
             'struct_like_set_test.cc',

src/iceberg/test/retry_util_test.cc

@@ -0,0 +1,304 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/util/retry_util.h"
+
+#include <chrono>
+#include <thread>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/result.h"
+#include "iceberg/test/matchers.h"
+
+namespace iceberg {
+
+TEST(RetryRunnerTest, SuccessOnFirstAttempt) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = RetryRunner(RetryConfig{.num_retries = 3,
+                                        .min_wait_ms = 1,
+                                        .max_wait_ms = 10,
+                                        .total_timeout_ms = 5000})
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          return 42;
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsOk());
+  EXPECT_EQ(*result, 42);
+  EXPECT_EQ(call_count, 1);
+  EXPECT_EQ(attempts, 1);
+}
+
+TEST(RetryRunnerTest, RetryOnceThenSucceed) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = RetryRunner(RetryConfig{.num_retries = 3,
+                                        .min_wait_ms = 1,
+                                        .max_wait_ms = 10,
+                                        .total_timeout_ms = 5000})
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          if (call_count == 1) {
+                            return CommitFailed("transient failure");
+                          }
+                          return 42;
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsOk());
+  EXPECT_EQ(*result, 42);
+  EXPECT_EQ(call_count, 2);
+  EXPECT_EQ(attempts, 2);
+}
+
+TEST(RetryRunnerTest, MaxAttemptsExhausted) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = RetryRunner(RetryConfig{.num_retries = 2,
+                                        .min_wait_ms = 1,
+                                        .max_wait_ms = 10,
+                                        .total_timeout_ms = 5000})
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          return CommitFailed("always fails");
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsError(ErrorKind::kCommitFailed));
+  EXPECT_EQ(call_count, 3);
+  EXPECT_EQ(attempts, 3);
+}
+
+TEST(RetryRunnerTest, OnlyRetryOnFilter) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = RetryRunner(RetryConfig{.num_retries = 3,
+                                        .min_wait_ms = 1,
+                                        .max_wait_ms = 10,
+                                        .total_timeout_ms = 5000})
+                    .OnlyRetryOn(ErrorKind::kCommitFailed)
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          return ValidationFailed("schema conflict");
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+  EXPECT_EQ(call_count, 1);
+  EXPECT_EQ(attempts, 1);
+}
+
+TEST(RetryRunnerTest, OnlyRetryOnMatchingError) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = RetryRunner(RetryConfig{.num_retries = 2,
+                                        .min_wait_ms = 1,
+                                        .max_wait_ms = 10,
+                                        .total_timeout_ms = 5000})
+                    .OnlyRetryOn(ErrorKind::kCommitFailed)
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          if (call_count <= 2) {
+                            return CommitFailed("transient");
+                          }
+                          return 100;
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsOk());
+  EXPECT_EQ(*result, 100);
+  EXPECT_EQ(call_count, 3);
+  EXPECT_EQ(attempts, 3);
+}
+
+TEST(RetryRunnerTest, StopRetryOnMatchingError) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = RetryRunner(RetryConfig{.num_retries = 5,
+                                        .min_wait_ms = 1,
+                                        .max_wait_ms = 10,
+                                        .total_timeout_ms = 5000})
+                    .StopRetryOn({ErrorKind::kCommitStateUnknown})
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          return CommitStateUnknown("datacenter on fire");
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsError(ErrorKind::kCommitStateUnknown));
+  EXPECT_EQ(call_count, 1);
+  EXPECT_EQ(attempts, 1);
+}
+
+TEST(RetryRunnerTest, ZeroRetries) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = RetryRunner(RetryConfig{.num_retries = 0,
+                                        .min_wait_ms = 1,
+                                        .max_wait_ms = 10,
+                                        .total_timeout_ms = 5000})
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          return CommitFailed("fail");
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsError(ErrorKind::kCommitFailed));
+  EXPECT_EQ(call_count, 1);
+  EXPECT_EQ(attempts, 1);
+}
+
+TEST(RetryRunnerTest, TotalTimeoutStopsBeforeStartingAnotherAttempt) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = RetryRunner(RetryConfig{.num_retries = 3,
+                                        .min_wait_ms = 20,
+                                        .max_wait_ms = 20,
+                                        .total_timeout_ms = 15})
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          // The first failure consumes most of the 15 ms budget, so the
+                          // next 20 ms backoff should prevent another attempt from
+                          // starting.
+                          if (call_count == 1) {
+                            std::this_thread::sleep_for(std::chrono::milliseconds(10));
+                          }
+                          return CommitFailed("retry budget exhausted");
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsError(ErrorKind::kCommitFailed));
+  EXPECT_EQ(call_count, 1);
+  EXPECT_EQ(attempts, 1);
+}
+
+TEST(RetryRunnerTest, MakeCommitRetryRunnerConfig) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = MakeCommitRetryRunner(2, 1, 10, 5000)
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          return ValidationFailed("not retryable");
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+  EXPECT_EQ(call_count, 1);
+  EXPECT_EQ(attempts, 1);
+}
+
+TEST(RetryRunnerTest, MakeCommitRetryRunnerRetriesCommitFailed) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = MakeCommitRetryRunner(3, 1, 10, 5000)
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          if (call_count <= 2) {
+                            return CommitFailed("transient");
+                          }
+                          return 99;
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsOk());
+  EXPECT_EQ(*result, 99);
+  EXPECT_EQ(call_count, 3);
+  EXPECT_EQ(attempts, 3);
+}
+
+TEST(RetryRunnerTest, OnlyRetryOnMultipleErrorKinds) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result =
+      RetryRunner(RetryConfig{.num_retries = 5,
+                              .min_wait_ms = 1,
+                              .max_wait_ms = 10,
+                              .total_timeout_ms = 5000})
+          .OnlyRetryOn({ErrorKind::kCommitFailed, ErrorKind::kServiceUnavailable})
+          .Run(
+              [&]() -> Result<int> {
+                ++call_count;
+                if (call_count == 1) {
+                  return CommitFailed("conflict");
+                }
+                if (call_count == 2) {
+                  return ServiceUnavailable("server busy");
+                }
+                return 77;
+              },
+              &attempts);
+
+  EXPECT_THAT(result, IsOk());
+  EXPECT_EQ(*result, 77);
+  EXPECT_EQ(call_count, 3);
+  EXPECT_EQ(attempts, 3);
+}
+
+TEST(RetryRunnerTest, DefaultRetryAllErrors) {
+  int call_count = 0;
+  int32_t attempts = 0;
+
+  auto result = RetryRunner(RetryConfig{.num_retries = 3,
+                                        .min_wait_ms = 1,
+                                        .max_wait_ms = 10,
+                                        .total_timeout_ms = 5000})
+                    .Run(
+                        [&]() -> Result<int> {
+                          ++call_count;
+                          if (call_count == 1) {
+                            return IOError("disk full");
+                          }
+                          if (call_count == 2) {
+                            return ValidationFailed("bad schema");
+                          }
+                          return 55;
+                        },
+                        &attempts);
+
+  EXPECT_THAT(result, IsOk());
+  EXPECT_EQ(*result, 55);
+  EXPECT_EQ(call_count, 3);
+  EXPECT_EQ(attempts, 3);
+}
+
+}  // namespace iceberg

src/iceberg/test/table_test.cc

@@ -128,6 +128,9 @@ TYPED_TEST(TypedTableTest, Refresh) {
           .WillOnce(::testing::Return(refreshed));
     }
     EXPECT_THAT(table->Refresh(), IsOk());
+    if constexpr (std::is_same_v<TypeParam, Table>) {
+      EXPECT_EQ(table->metadata_file_location(), "s3://bucket/meta2.json");
+    }
   } else {
     EXPECT_THAT(table->Refresh(), IsError(ErrorKind::kNotSupported));
   }