Выполнение повторных попыток

Стандартный пакет database/sql использует внутреннюю логику выполнения повторов на основе тех ошибок, что возвращает конкретная реализация драйвера.
Так, в коде пакета database/sql во многих местах можно встретить политику повторов из трех попыток:

• 2 попытки выполнить на существующем соединении или новом (если пул соединений database/sql пуст)
• 1 попытка выполнить на новом соединении.

В большинстве случаев такой политики повторов достаточно, чтобы пережить временную недоступность нод YDB или проблемы с сессией YDB.

YDB Go SDK предоставляет специальные функции для гарантированного выполнения пользовательской операции:

import (
    "context"
    "database/sql"
    "fmt"
    "log"

    "github.com/ydb-platform/ydb-go-sdk/v3/retry"
)

func main() {
    ...
    err = retry.Do(ctx, db, func(ctx context.Context, cc *sql.Conn) (err error) {
        row = cc.QueryRowContext(ctx, `
                PRAGMA TablePathPrefix("/local");
                DECLARE $seriesID AS Uint64;
                DECLARE $seasonID AS Uint64;
                DECLARE $episodeID AS Uint64;
                SELECT views FROM episodes WHERE series_id = $seriesID AND season_id = $seasonID AND episode_id = $episodeID;
            `,
            sql.Named("seriesID", uint64(1)),
            sql.Named("seasonID", uint64(1)),
            sql.Named("episodeID", uint64(1)),
        )
        var views sql.NullFloat64
        if err = row.Scan(&views); err != nil {
            return fmt.Errorf("cannot scan views: %w", err)
        }
        if views.Valid {
            return fmt.Errorf("unexpected valid views: %v", views.Float64)
        }
        log.Printf("views = %v", views)
        return row.Err()
    }, retry.WithDoRetryOptions(retry.WithIdempotent(true)))
    if err != nil {
        log.Printf("retry.Do failed: %v\n", err)
    }
}

import (
    "context"
    "database/sql"
    "fmt"
    "log"

    "github.com/ydb-platform/ydb-go-sdk/v3/retry"
)

func main() {
    ...
    err = retry.DoTx(ctx, db, func(ctx context.Context, tx *sql.Tx) error {
        row := tx.QueryRowContext(ctx,`
                PRAGMA TablePathPrefix("/local");
                DECLARE $seriesID AS Uint64;
                DECLARE $seasonID AS Uint64;
                DECLARE $episodeID AS Uint64;
                SELECT views FROM episodes WHERE series_id = $seriesID AND season_id = $seasonID AND episode_id = $episodeID;
            `,
            sql.Named("seriesID", uint64(1)),
            sql.Named("seasonID", uint64(1)),
            sql.Named("episodeID", uint64(1)),
        )
        var views sql.NullFloat64
        if err = row.Scan(&views); err != nil {
            return fmt.Errorf("cannot select current views: %w", err)
        }
        if !views.Valid {
            return fmt.Errorf("unexpected invalid views: %v", views)
        }
        t.Logf("views = %v", views)
        if views.Float64 != 1 {
            return fmt.Errorf("unexpected views value: %v", views)
        }
        return nil
    }, retry.WithDoTxRetryOptions(retry.WithIdempotent(true)), retry.WithTxOptions(&sql.TxOptions{
        Isolation: sql.LevelSnapshot,
        ReadOnly:  true,
    }))
    if err != nil {
        log.Printf("do tx failed: %v\n", err)
    }
}

В YDB Java SDK механизм повторных запросов реализован в виде класс хелпера com.yandex.ydb.table.SessionRetryContext. Данный класс конструируется с помощью метода SessionRetryContext.create в который требуется передать реализацию интерфейса SessionSupplier - как правило это экземпляр класса TableClient.
Дополнительно пользователь может задавать некоторые другие опции

• maxRetries(int maxRetries) - максимальное количество повторов операции, не включает в себя первое выполение. Значение по умолчанию 10
• retryNotFound(boolean retryNotFound) - опция повтора операций, вернувших статус NOT_FOUND. По умолчанию включено.
• idempotent(boolean idempotent) - признак идемпотентности операций. Идемпотентные операции будут повторяться для более широкого списка ошибок. По умолчанию отключено.

Для запуска операций с ретраями класс SessionRetryContext предоставляет два метода

• CompletableFuture<Status> supplyStatus - выполнение операции, возвращающей статус. В качестве аргумента принимает лямбду Function<Session, CompletableFuture<Status>> fn
• CompletableFuture<Result<T>> supplyResult - выполнение операции, возвращающей данные. В качестве аргумента принимает лямбду Function<Session, CompletableFuture<Result<T>>> fn

При использовании класса SessionRetryContext нужно учитывать, что повторое исполнение операции будут выполнятся в следующих случаях

• Лямбда вернула retryable код ошибки

• В рамках исполнения лямбды была вызвано UnexpectedResultException c retryable кодом ошибки

  Пример кода, использующего SessionRetryContext.supplyStatus:

  private void createTable(TableClient tableClient, String database, String tableName) {
      SessionRetryContext retryCtx = SessionRetryContext.create(tableClient).build();
      TableDescription pets = TableDescription.newBuilder()
              .addNullableColumn("species", PrimitiveType.Text)
              .addNullableColumn("name", PrimitiveType.Text)
              .addNullableColumn("color", PrimitiveType.Text)
              .addNullableColumn("price", PrimitiveType.Float)
              .setPrimaryKeys("species", "name")
              .build();
  
      String tablePath = database + "/" + tableName;
      retryCtx.supplyStatus(session -> session.createTable(tablePath, pets))
              .join().expectSuccess();
  }
  

  Пример кода, использующего SessionRetryContext.supplyResult:

  private void selectData(TableClient tableClient, String tableName) {
      SessionRetryContext retryCtx = SessionRetryContext.create(tableClient).build();
      String selectQuery
              = "DECLARE $species AS Text;"
              + "DECLARE $name AS Text;"
              + "SELECT * FROM " + tableName + " "
              + "WHERE species = $species AND name = $name;";
  
      Params params = Params.of(
              "$species", PrimitiveValue.newText("cat"),
              "$name", PrimitiveValue.newText("Tom")
      );
  
      DataQueryResult data = retryCtx
              .supplyResult(session -> session.executeDataQuery(selectQuery, TxControl.onlineRo(), params))
              .join().getValue();
  
      ResultSetReader rsReader = data.getResultSet(0);
      logger.info("Result of select query:");
      while (rsReader.next()) {
          logger.info("  species: {}, name: {}, color: {}, price: {}",
                  rsReader.getColumn("species").getText(),
                  rsReader.getColumn("name").getText(),
                  rsReader.getColumn("color").getText(),
                  rsReader.getColumn("price").getFloat()
          );
      }
  }