9 replies

Transactions inside an AshOban worker

Hi, I'm working with AshOban and I saw that my

after_action

after_action

after_action

after_action failed, which I thought would cause the change on my resource to roll back, but it didn't.
I might just have understood it wrong, but I thought that would only happen if one were using

after_transaction

after_transaction

after_transaction

after_transaction, since the

after_action

after_action

after_action

after_action runs inside the same transaction as the main change right?

For reference, it was running inside an AshOban worker and I'm using SQLite as my datalayer - maybe that has something to do with it?

It happened when an AshOban worker was running the

:create_openai_batch

:create_openai_batch

:create_openai_batch

:create_openai_batch action. Both resources:

Batch

Batch

Batch

Batch and

Request

Request

Request

Request as well as the stdout is attached.

Below is the change performed in the

:create_openai_batch

:create_openai_batch

:create_openai_batch

:create_openai_batch:

defmodule Batcher.Batching.Changes.CreateOpenaiBatch do
  use Ash.Resource.Change
  require Logger

  alias Batcher.{OpenaiApiClient}

  @impl true
  def change(changeset, _opts, _context) do
    batch = changeset.data

    changeset
    |> Ash.Changeset.before_transaction(fn changeset ->
      # Create batch on OpenAI before transaction starts in case it fails
      case OpenaiApiClient.create_batch(batch.openai_input_file_id, batch.url) do
        {:ok, response} ->
          changeset
          |> Ash.Changeset.force_change_attribute(:openai_batch_id, response["id"])

        {:error, reason} ->
          Ash.Changeset.add_error(changeset, "OpenAI batch creation failed: #{reason}")
      end
    end)
    |> Ash.Changeset.after_action(fn _changeset, batch ->
      # Bulk update all pending requests to processing after transaction
      Batcher.Batching.Request
      |> Ash.Query.filter(batch_id == ^batch.id)
      |> Ash.bulk_update!(:begin_processing, %{},
        strategy: :stream, # Made it work with stream, the error ocurred without this specified
        return_errors?: true
      )

      {:ok, batch}
    end)
  end
end

defmodule Batcher.Batching.Changes.CreateOpenaiBatch do
  use Ash.Resource.Change
  require Logger

  alias Batcher.{OpenaiApiClient}

  @impl true
  def change(changeset, _opts, _context) do
    batch = changeset.data

    changeset
    |> Ash.Changeset.before_transaction(fn changeset ->
      # Create batch on OpenAI before transaction starts in case it fails
      case OpenaiApiClient.create_batch(batch.openai_input_file_id, batch.url) do
        {:ok, response} ->
          changeset
          |> Ash.Changeset.force_change_attribute(:openai_batch_id, response["id"])

        {:error, reason} ->
          Ash.Changeset.add_error(changeset, "OpenAI batch creation failed: #{reason}")
      end
    end)
    |> Ash.Changeset.after_action(fn _changeset, batch ->
      # Bulk update all pending requests to processing after transaction
      Batcher.Batching.Request
      |> Ash.Query.filter(batch_id == ^batch.id)
      |> Ash.bulk_update!(:begin_processing, %{},
        strategy: :stream, # Made it work with stream, the error ocurred without this specified
        return_errors?: true
      )

      {:ok, batch}
    end)
  end
end

defmodule Batcher.Batching.Changes.CreateOpenaiBatch do
  use Ash.Resource.Change
  require Logger

  alias Batcher.{OpenaiApiClient}

  @impl true
  def change(changeset, _opts, _context) do
    batch = changeset.data

    changeset
    |> Ash.Changeset.before_transaction(fn changeset ->
      # Create batch on OpenAI before transaction starts in case it fails
      case OpenaiApiClient.create_batch(batch.openai_input_file_id, batch.url) do
        {:ok, response} ->
          changeset
          |> Ash.Changeset.force_change_attribute(:openai_batch_id, response["id"])

        {:error, reason} ->
          Ash.Changeset.add_error(changeset, "OpenAI batch creation failed: #{reason}")
      end
    end)
    |> Ash.Changeset.after_action(fn _changeset, batch ->
      # Bulk update all pending requests to processing after transaction
      Batcher.Batching.Request
      |> Ash.Query.filter(batch_id == ^batch.id)
      |> Ash.bulk_update!(:begin_processing, %{},
        strategy: :stream, # Made it work with stream, the error ocurred without this specified
        return_errors?: true
      )

      {:ok, batch}
    end)
  end
end

defmodule Batcher.Batching.Changes.CreateOpenaiBatch do
  use Ash.Resource.Change
  require Logger

  alias Batcher.{OpenaiApiClient}

  @impl true
  def change(changeset, _opts, _context) do
    batch = changeset.data

    changeset
    |> Ash.Changeset.before_transaction(fn changeset ->
      # Create batch on OpenAI before transaction starts in case it fails
      case OpenaiApiClient.create_batch(batch.openai_input_file_id, batch.url) do
        {:ok, response} ->
          changeset
          |> Ash.Changeset.force_change_attribute(:openai_batch_id, response["id"])

        {:error, reason} ->
          Ash.Changeset.add_error(changeset, "OpenAI batch creation failed: #{reason}")
      end
    end)
    |> Ash.Changeset.after_action(fn _changeset, batch ->
      # Bulk update all pending requests to processing after transaction
      Batcher.Batching.Request
      |> Ash.Query.filter(batch_id == ^batch.id)
      |> Ash.bulk_update!(:begin_processing, %{},
        strategy: :stream, # Made it work with stream, the error ocurred without this specified
        return_errors?: true
      )

      {:ok, batch}
    end)
  end
end

Any help is much appreciated as I'm at the steep part of the Ash learning curve atm

message.txt8.27KB

message.txt6.19KB

message.txt4.88KB

Transactions inside an AshOban worker

Similar Threads

Similar Threads

Similar Threads