Transacciones de Cloud Datastore

Una transacción es un conjunto de operaciones de Datastore en una o más entidades, en un máximo de 25 grupos de entidades. Se garantiza que toda transacción es atómica, es decir, que nunca se aplica de forma parcial. Se aplican todas las operaciones de la transacción o ninguna de estas.

Usa transacciones

Las transacciones tienen una duración máxima de 270 segundos, con un tiempo de vencimiento por inactividad de 10 segundos después de 30 segundos.

Las operaciones pueden fallar cuando ocurren las siguientes situaciones:

Se intentan aplicar demasiadas modificaciones simultáneas en el mismo grupo de entidad.
La transacción supera un límite de recursos.
Se produce un error interno en Datastore.

En todos estos casos, la API de Datastore muestra un error.

Las transacciones son una función opcional de Datastore; no es necesario que las uses para realizar operaciones de Datastore.

Una aplicación puede ejecutar un conjunto de instrucciones y operaciones de Datastore en una sola transacción, de modo que, si alguna operación o instrucción genera una excepción, no se aplica ninguna de las operaciones de Datastore del conjunto. La aplicación define las acciones que se deben realizar en la transacción.

El siguiente fragmento muestra cómo realizar una transacción con la API de Datastore. Transfiere dinero de una cuenta a otra.

C#

Para obtener información sobre cómo instalar y usar la biblioteca cliente de Cloud Datastore, consulta Bibliotecas cliente de Cloud Datastore. Si deseas obtener más información, consulta la documentación de referencia de la API de C# de Cloud Datastore.

Para autenticarte en Cloud Datastore, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local.

private void TransferFunds(Key fromKey, Key toKey, long amount)
{
    using (var transaction = _db.BeginTransaction())
    {
        var entities = transaction.Lookup(fromKey, toKey);
        entities[0]["balance"].IntegerValue -= amount;
        entities[1]["balance"].IntegerValue += amount;
        transaction.Update(entities);
        transaction.Commit();
    }
}

Go

type BankAccount struct {
	Balance int
}

const amount = 50
keys := []*datastore.Key{to, from}
tx, err := client.NewTransaction(ctx)
if err != nil {
	log.Fatalf("client.NewTransaction: %v", err)
}
accs := make([]BankAccount, 2)
if err := tx.GetMulti(keys, accs); err != nil {
	tx.Rollback()
	log.Fatalf("tx.GetMulti: %v", err)
}
accs[0].Balance += amount
accs[1].Balance -= amount
if _, err := tx.PutMulti(keys, accs); err != nil {
	tx.Rollback()
	log.Fatalf("tx.PutMulti: %v", err)
}
if _, err = tx.Commit(); err != nil {
	log.Fatalf("tx.Commit: %v", err)
}

Java

void transferFunds(Key fromKey, Key toKey, long amount) {
  Transaction txn = datastore.newTransaction();
  try {
    List<Entity> entities = txn.fetch(fromKey, toKey);
    Entity from = entities.get(0);
    Entity updatedFrom =
        Entity.newBuilder(from).set("balance", from.getLong("balance") - amount).build();
    Entity to = entities.get(1);
    Entity updatedTo =
        Entity.newBuilder(to).set("balance", to.getLong("balance") + amount).build();
    txn.put(updatedFrom, updatedTo);
    txn.commit();
  } finally {
    if (txn.isActive()) {
      txn.rollback();
    }
  }
}

Node.js

async function transferFunds(fromKey, toKey, amount) {
  const transaction = datastore.transaction();
  await transaction.run();
  const results = await Promise.all([
    transaction.get(fromKey),
    transaction.get(toKey),
  ]);
  const accounts = results.map(result => result[0]);

  accounts[0].balance -= amount;
  accounts[1].balance += amount;

  transaction.save([
    {
      key: fromKey,
      data: accounts[0],
    },
    {
      key: toKey,
      data: accounts[1],
    },
  ]);

  return await transaction.commit();
}

PHP

/**
 * Update two entities in a transaction.
 *
 * @param DatastoreClient $datastore
 * @param Key $fromKey
 * @param Key $toKey
 * @param $amount
 */
function transfer_funds(
    DatastoreClient $datastore,
    Key $fromKey,
    Key $toKey,
    $amount
) {
    $transaction = $datastore->transaction();
    // The option 'sort' is important here, otherwise the order of the result
    // might be different from the order of the keys.
    $result = $transaction->lookupBatch([$fromKey, $toKey], ['sort' => true]);
    if (count($result['found']) != 2) {
        $transaction->rollback();
    }
    $fromAccount = $result['found'][0];
    $toAccount = $result['found'][1];
    $fromAccount['balance'] -= $amount;
    $toAccount['balance'] += $amount;
    $transaction->updateBatch([$fromAccount, $toAccount]);
    $transaction->commit();
}

Python

from google.cloud import datastore

# For help authenticating your client, visit
# https://cloud.google.com/docs/authentication/getting-started
client = datastore.Client()

def transfer_funds(client, from_key, to_key, amount):
    with client.transaction():
        from_account = client.get(from_key)
        to_account = client.get(to_key)

        from_account["balance"] -= amount
        to_account["balance"] += amount

        client.put_multi([from_account, to_account])

Ruby

def transfer_funds from_key, to_key, amount
  datastore.transaction do |tx|
    from = tx.find from_key
    from["balance"] -= amount
    to = tx.find to_key
    to["balance"] += amount
    tx.save from, to
  end
end

Ten en cuenta que, para que nuestros ejemplos sean más concisos, a veces omitimos rollback si la transacción falla. En el código de producción, es importante asegurarse de que todas las transacciones se confirmen o reviertan de manera explícita.

¿Qué se puede hacer en una transacción?

Todas las operaciones de Datastore en una transacción pueden operar en un máximo de veinticinco grupos de entidad. Esto incluye las consultas de entidades por entidad principal, la recuperación de entidades por clave y la actualización y eliminación de entidades.

Cuando dos o más transacciones intentan modificar entidades de forma simultánea en uno o más grupos de entidad comunes, solo la primera transacción que confirme los cambios se realizará correctamente, y la confirmación de las demás fallará. Debido a este diseño, el uso de grupos de entidad limita el número de escrituras simultáneas que puedes realizar en cualquier entidad de los grupos. Cuando se inicia una transacción, Datastore utiliza el control de simultaneidad optimista mediante la comprobación de la última hora de actualización de los grupos de entidad utilizados en la transacción. Luego de confirmar una transacción para los grupos de entidad, Datastore verifica nuevamente la última hora de actualización de los grupos de entidad utilizados en la transacción. Si la hora cambió desde la verificación inicial, se muestra un error. Para obtener una explicación sobre los grupos de entidad, consulta Rutas de las entidades principales.

Aislamiento y coherencia

Fuera de las transacciones, el nivel de aislamiento de Datastore es el más cercano al de lectura confirmada. Dentro de las transacciones, se aplica el aislamiento serializable. es decir, una transacción no puede modificar de forma simultánea los datos que lee o modifica otra que ocurrió primero. Lee la wiki sobre el aislamiento serializable y el artículo Aislamiento de transacción para obtener más información sobre los niveles de aislamiento.

En una transacción, todas las lecturas reflejan el estado coherente actual de Datastore en el momento en que comenzó la transacción. Se garantiza que las consultas y las búsquedas dentro de una transacción verán una sola instantánea coherente de Datastore desde el comienzo de la transacción. Las filas de las entidades y de los índices en los grupos de entidades de la transacción se actualizan por completo para que las consultas muestren el conjunto completo y correcto de las entidades del resultado, sin los falsos positivos ni falsos negativos que se describen en Aislamiento de transacción y que pueden generarse en las consultas realizadas fuera de las transacciones.

Esta vista de instantánea coherente también se extiende a las lecturas realizadas luego de las escrituras dentro de las transacciones. A diferencia de la mayoría de las bases de datos, las consultas y las operaciones get en una transacción de Datastore no ven los resultados de las escrituras anteriores dentro de esa transacción. En particular, si se modifica o borra una entidad en una transacción, una consulta o búsqueda muestra la versión original de la entidad, como estaba al principio de la transacción, o no muestra nada si la entidad no existía en ese momento.

Usos de las transacciones

Uno de los usos de las transacciones es la actualización de una entidad con un valor de propiedad nuevo relativo a su valor actual. El ejemplo de transferFunds anterior hace eso para dos entidades mediante la extracción de dinero de una cuenta y su transferencia a otra. La API de Datastore no reintenta transacciones de forma automática, pero puedes agregar tu propia lógica para reintentarlas, por ejemplo, a fin de manejar conflictos cuando otra solicitud actualice la misma entidad al mismo tiempo.

C#

/// <summary>
/// Retry the action when a Grpc.Core.RpcException is thrown.
/// </summary>
private T RetryRpc<T>(Func<T> action)
{
    List<Grpc.Core.RpcException> exceptions = null;
    var delayMs = _retryDelayMs;
    for (int tryCount = 0; tryCount < _retryCount; ++tryCount)
    {
        try
        {
            return action();
        }
        catch (Grpc.Core.RpcException e)
        {
            if (exceptions == null)
                exceptions = new List<Grpc.Core.RpcException>();
            exceptions.Add(e);
        }
        System.Threading.Thread.Sleep(delayMs);
        delayMs *= 2;  // Exponential back-off.
    }
    throw new AggregateException(exceptions);
}

private void RetryRpc(Action action)
{
    RetryRpc(() => { action(); return 0; });
}

[Fact]
public void TestTransactionalRetry()
{
    int tryCount = 0;
    var keys = UpsertBalances();
    RetryRpc(() =>
    {
        using (var transaction = _db.BeginTransaction())
        {
            TransferFunds(keys[0], keys[1], 10, transaction);
            // Insert a conflicting transaction on the first try.
            if (tryCount++ == 0)
                TransferFunds(keys[1], keys[0], 5);
            transaction.Commit();
        }
    });
    Assert.Equal(2, tryCount);
}

Go

type BankAccount struct {
	Balance int
}

const amount = 50
_, err := client.RunInTransaction(ctx, func(tx *datastore.Transaction) error {
	keys := []*datastore.Key{to, from}
	accs := make([]BankAccount, 2)
	if err := tx.GetMulti(keys, accs); err != nil {
		return err
	}
	accs[0].Balance += amount
	accs[1].Balance -= amount
	_, err := tx.PutMulti(keys, accs)
	return err
})

Java

int retries = 5;
while (true) {
  try {
    transferFunds(fromKey, toKey, 10);
    break;
  } catch (DatastoreException e) {
    if (retries == 0) {
      throw e;
    }
    --retries;
  }
}
// Retry handling can also be configured and automatically applied using google-cloud-java.

Node.js

async function transferFundsWithRetry() {
  const maxTries = 5;

  async function tryRequest(currentAttempt, delay) {
    try {
      await transferFunds(fromKey, toKey, 10);
    } catch (err) {
      if (currentAttempt <= maxTries) {
        // Use exponential backoff
        setTimeout(async () => {
          await tryRequest(currentAttempt + 1, delay * 2);
        }, delay);
      }
      throw err;
    }
  }

  await tryRequest(1, 100);
}

PHP

$retries = 5;
for ($i = 0; $i < $retries; $i++) {
    try {
        transfer_funds($datastore, $fromKey, $toKey, 10);
    } catch (\Google\Cloud\Core\Exception\ConflictException $e) {
        // if $i >= $retries, the failure is final
        continue;
    }
    // Succeeded!
    break;
}

Python

from google.cloud import datastore

# For help authenticating your client, visit
# https://cloud.google.com/docs/authentication/getting-started
client = datastore.Client()

for _ in range(5):
    try:
        transfer_funds(client, account1.key, account2.key, 50)
        break
    except google.cloud.exceptions.Conflict:
        continue
else:
    print("Transaction failed.")

Ruby

(1..5).each do |i|
  begin
    return transfer_funds from_key, to_key, amount
  rescue Google::Cloud::Error => e
    raise e if i == 5
  end
end

Esto requiere una transacción porque otro usuario podría actualizar el valor de balance en una entidad después de que este código recupere el objeto, pero antes de que lo guarde modificado. Sin una transacción, la solicitud del usuario usa el valor de balance antes de la actualización del otro usuario, y el guardado reemplaza el valor nuevo. Mediante una transacción, la app se entera de la actualización del otro usuario.

Otro uso común de las transacciones es recuperar una entidad con una clave con nombre o crearla si aún no existe (este ejemplo se basa en el de listas de tareas de Crea una entidad):

C#

Entity task;
using (var transaction = _db.BeginTransaction())
{
    task = transaction.Lookup(_sampleTask.Key);
    if (task == null)
    {
        transaction.Insert(_sampleTask);
        transaction.Commit();
    }
}

Go

_, err := client.RunInTransaction(ctx, func(tx *datastore.Transaction) error {
	var task Task
	if err := tx.Get(key, &task); err != datastore.ErrNoSuchEntity {
		return err
	}
	_, err := tx.Put(key, &Task{
		Category:    "Personal",
		Done:        false,
		Priority:    4,
		Description: "Learn Cloud Datastore",
	})
	return err
})

Java

Entity task;
Transaction txn = datastore.newTransaction();
try {
  task = txn.get(taskKey);
  if (task == null) {
    task = Entity.newBuilder(taskKey).build();
    txn.put(task);
    txn.commit();
  }
} finally {
  if (txn.isActive()) {
    txn.rollback();
  }
}

Node.js

async function getOrCreate(taskKey, taskData) {
  const taskEntity = {
    key: taskKey,
    data: taskData,
  };
  const transaction = datastore.transaction();

  try {
    await transaction.run();
    const [task] = await transaction.get(taskKey);
    if (task) {
      // The task entity already exists.
      await transaction.rollback();
    } else {
      // Create the task entity.
      transaction.save(taskEntity);
      await transaction.commit();
    }
    return taskEntity;
  } catch (err) {
    await transaction.rollback();
  }
}

PHP

$transaction = $datastore->transaction();
$entity = $transaction->lookup($task->key());
if ($entity === null) {
    $entity = $transaction->insert($task);
    $transaction->commit();
}

Python

from google.cloud import datastore

# For help authenticating your client, visit
# https://cloud.google.com/docs/authentication/getting-started
client = datastore.Client()

with client.transaction():
    key = client.key(
        "Task", datetime.datetime.now(tz=datetime.timezone.utc).isoformat()
    )

    task = client.get(key)

    if not task:
        task = datastore.Entity(key)
        task.update({"description": "Example task"})
        client.put(task)

    return task

Ruby

task = nil
datastore.transaction do |tx|
  task = tx.find task_key
  if task.nil?
    task = datastore.entity task_key do |t|
      t["category"] = "Personal"
      t["done"] = false
      t["priority"] = 4
      t["description"] = "Learn Cloud Datastore"
    end
    tx.save task
  end
end

Al igual que antes, se necesita una transacción para resolver el caso en el que otro usuario intenta crear o actualizar una entidad con el mismo ID de string. Sin una transacción, si la entidad no existe y dos usuarios intentan crearla, el segundo reemplazará los datos del primero sin saberlo.

Cuando una transacción falla, puedes hacer que tu app vuelva a intentarla hasta que tenga éxito, o bien puedes dejar que tus usuarios manejen el error. Para ello, deberás propagarlo al nivel de la interfaz de usuario de la app. No es necesario crear un bucle de reintento en cada transacción.

Por último, puedes usar una transacción para leer una instantánea coherente de Datastore. Esto puede ser útil cuando se necesitan varias lecturas a fin de procesar una página o exportar datos que deben ser coherentes. Estos tipos de transacciones suelen llamarse de solo lectura, ya que no realizan ninguna escritura. Las transacciones de solo lectura que se realizan en un solo grupo no fallan nunca debido a modificaciones simultáneas, por lo que no debes implementar reintentos en caso de fallas. Sin embargo, las transacciones que se realizan en varios grupos de entidades pueden fallar debido a modificaciones simultáneas, por lo que deben tener reintentos.

C#

Entity taskList;
IReadOnlyList<Entity> tasks;
using (var transaction = _db.BeginTransaction(TransactionOptions.CreateReadOnly()))
{
    taskList = transaction.Lookup(taskListKey);
    var query = new Query("Task")
    {
        Filter = Filter.HasAncestor(taskListKey)
    };
    tasks = transaction.RunQuery(query).Entities;
    transaction.Commit();
}

Go

tx, err := client.NewTransaction(ctx, datastore.ReadOnly)
if err != nil {
	log.Fatalf("client.NewTransaction: %v", err)
}
defer tx.Rollback() // Transaction only used for read.

ancestor := datastore.NameKey("TaskList", "default", nil)
query := datastore.NewQuery("Task").Ancestor(ancestor).Transaction(tx)
var tasks []Task
_, err = client.GetAll(ctx, query, &tasks)

Java

Entity taskList;
QueryResults<Entity> tasks;
Transaction txn =
    datastore.newTransaction(
        TransactionOptions.newBuilder().setReadOnly(ReadOnly.newBuilder().build()).build());
try {
  taskList = txn.get(taskListKey);
  Query<Entity> query =
      Query.newEntityQueryBuilder()
          .setKind("Task")
          .setFilter(PropertyFilter.hasAncestor(taskListKey))
          .build();
  tasks = txn.run(query);
  txn.commit();
} finally {
  if (txn.isActive()) {
    txn.rollback();
  }
}

Node.js

async function getTaskListEntities() {
  const transaction = datastore.transaction({readOnly: true});
  try {
    const taskListKey = datastore.key(['TaskList', 'default']);

    await transaction.run();
    const [taskList] = await transaction.get(taskListKey);
    const query = datastore.createQuery('Task').hasAncestor(taskListKey);
    const [taskListEntities] = await transaction.runQuery(query);
    await transaction.commit();
    return [taskList, taskListEntities];
  } catch (err) {
    await transaction.rollback();
  }
}

PHP

$transaction = $datastore->readOnlyTransaction();
$taskListKey = $datastore->key('TaskList', 'default');
$query = $datastore->query()
    ->kind('Task')
    ->hasAncestor($taskListKey);
$result = $transaction->runQuery($query);
$taskListEntities = [];
$num = 0;
/* @var Entity $task */
foreach ($result as $task) {
    $taskListEntities[] = $task;
    $num += 1;
}

Python

from google.cloud import datastore

# For help authenticating your client, visit
# https://cloud.google.com/docs/authentication/getting-started
client = datastore.Client()

with client.transaction(read_only=True):
    task_list_key = client.key("TaskList", "default")

    task_list = client.get(task_list_key)

    query = client.query(kind="Task", ancestor=task_list_key)
    tasks_in_list = list(query.fetch())

    return task_list, tasks_in_list

Ruby

# task_list_name = "default"
task_list_key = datastore.key "TaskList", task_list_name
datastore.read_only_transaction do |tx|
  task_list = tx.find task_list_key
  query = datastore.query("Task").ancestor(task_list)
  tasks_in_list = tx.run query
end

Transacciones y grupos de entidades

Un grupo de entidad es un conjunto de entidades conectadas de manera ascendente con un elemento raíz común. La organización de los datos dentro de los grupos de entidad puede limitar de esta manera las transacciones que se pueden realizar:

Todos los datos a los que accede una transacción deben estar contenidos en 25 grupos de entidades como máximo.
Si quieres usar consultas dentro de una transacción, los datos deben estar organizados en grupos de entidades de manera tal que puedas especificar filtros de principales que coincidan con los datos correctos.
Existe un límite de capacidad de procesamiento de escrituras de alrededor de una transacción por segundo dentro de un mismo grupo de entidades. Esta limitación existe porque Datastore realiza una replicación síncrona sin instancia principal de cada grupo de entidad en una gran área geográfica para brindar alta confiabilidad y tolerancia a errores.

Advertencia: Si se supera el límite de capacidad de procesamiento de escritura de una transacción por segundo dentro de un solo grupo de entidad, es posible que haya conflictos entre las operaciones de lectura y escritura en todos los grupos de entidad que participen en las transacciones.

En muchas aplicaciones, es aceptable usar la coherencia eventual (es decir, una consulta no principal que abarque varios grupos de entidades y que a veces puede mostrar datos ligeramente inactivos) cuando se obtiene una vista amplia de datos no relacionados y, luego, usar la coherencia sólida (una consulta principal o una lookup de una sola entidad) cuando se visualiza o edita un solo conjunto de datos muy relacionados. En tales aplicaciones, suele ser una buena estrategia usar un grupo de entidades separado para cada conjunto de datos sumamente relacionados. Para obtener más información, consulta Coherencia de datos.

¿Qué sigue?

Obtén más información sobre las consultas de Datastore.
Obtén más información sobre la coherencia de los datos en Datastore.