fix: handle new type constraints properly

lib/ash/actions/create/bulk.ex

@@ -21,14 +21,6 @@ defmodule Ash.Actions.Create.Bulk do
       raise ArgumentError, "Cannot specify `sorted?: true` and `return_stream?: true` together"
     end
 
-    opts =
-      if opts[:max_concurrency] && opts[:max_concurrency] > 0 &&
-           not Ash.DataLayer.can?(:async_engine, resource) do
-        Keyword.put(opts, :max_concurrency, 0)
-      else
-        opts
-      end
-
     if opts[:transaction] == :all &&
          Ash.DataLayer.data_layer_can?(resource, :transact) do
       notify? =
@@ -168,7 +160,7 @@ defmodule Ash.Actions.Create.Bulk do
         end,
         fn _ -> :ok end
       )
-      |> map_batches(opts, fn
+      |> map_batches(resource, opts, fn
         batch_config ->
           %{count: count, batch: batch, must_return_records?: must_return_records?} = batch_config
           context = batch |> Enum.at(0) |> Kernel.||(%{}) |> Map.get(:context)
@@ -497,9 +489,17 @@ defmodule Ash.Actions.Create.Bulk do
       %{result | errors: errors, error_count: error_count}
   end
 
-  defp map_batches(stream, opts, callback) do
+  defp map_batches(stream, resource, opts, callback) do
     max_concurrency = opts[:max_concurrency]
 
+    max_concurrency =
+      if max_concurrency && max_concurrency > 0 &&
+           not Ash.DataLayer.can?(:async_engine, resource) do
+        max_concurrency
+      else
+        0
+      end
+
     if max_concurrency && max_concurrency > 1 do
       Task.async_stream(
         stream,

lib/ash/type/datetime.ex

@@ -33,6 +33,7 @@ defmodule Ash.Type.DateTime do
   end
 
   @impl true
+  @spec storage_type(nonempty_maybe_improper_list()) :: any()
   def storage_type([{:storage_type, storage_type} | _]) do
     storage_type
   end

lib/ash/type/new_type.ex

@@ -127,7 +127,7 @@ defmodule Ash.Type.NewType do
 
       @impl Ash.Type
       def init(constraints) do
-        {:ok, type_constraints(constraints, unquote(subtype_constraints))}
+        unquote(subtype_of).init(type_constraints(constraints, unquote(subtype_constraints)))
       end
 
       @impl Ash.Type

lib/ash/type/type.ex

@@ -373,7 +373,10 @@ defmodule Ash.Type do
     end
   end
 
-  def init(type, constraints), do: type.init(constraints)
+  def init(type, constraints) do
+    type = get_type(type)
+    type.init(constraints)
+  end
 
   @doc """
   Returns the *underlying* storage type (the underlying type of the *ecto type* of the *ash type*)
@@ -946,7 +949,9 @@ defmodule Ash.Type do
         def init(opts) do
           constraints = @parent.constraints()
 
-          Keyword.take(opts, Keyword.keys(constraints))
+          opts
+          |> Keyword.take(Keyword.keys(constraints))
+          |> @parent.init()
         end
 
         @impl true

test/actions/bulk/bulk_create_test.exs

@@ -231,49 +231,6 @@ defmodule Ash.Test.Actions.BulkCreateTest do
              )
   end
 
-  test "can run batches concurrently" do
-    assert %Ash.BulkResult{
-             records: [
-               %{title: "title1", title2: "changes", title3: "wont"},
-               %{title: "title2", title2: "changes", title3: "wont"}
-             ]
-           } =
-             Api.bulk_create!(
-               [
-                 %{title: "title1", title2: "changes", title3: "wont"},
-                 %{title: "title2", title2: "changes", title3: "wont"}
-               ],
-               Post,
-               :create,
-               return_records?: true,
-               batch_size: 1,
-               max_concurrency: 2,
-               sorted?: true
-             )
-
-    assert %Ash.BulkResult{
-             records: [
-               %{title: "title1", title2: "did_change", title3: "wont"},
-               %{title: "title2", title2: "did_change", title3: "wont"}
-             ]
-           } =
-             Api.bulk_create!(
-               [
-                 %{title: "title1", title2: "did_change", title3: "oh no"},
-                 %{title: "title2", title2: "did_change", title3: "what happened"}
-               ],
-               Post,
-               :create,
-               return_records?: true,
-               batch_size: 1,
-               max_concurrency: 2,
-               upsert?: true,
-               upsert_identity: :unique_title,
-               upsert_fields: [:title2],
-               sorted?: true
-             )
-  end
-
   test "runs after action hooks" do
     assert %Ash.BulkResult{records: [%{title: "title1_stuff"}, %{title: "title2_stuff"}]} =
              Api.bulk_create!(

test/actions/load_test.exs

@@ -807,7 +807,7 @@ defmodule Ash.Test.Actions.LoadTest do
         |> new(%{name: "zerg"})
         |> Api.create!()
 
-      _post1 =
+      post1 =
         Post
         |> new(%{title: "post1"})
         |> manage_relationship(:author, author, type: :append_and_remove)
@@ -824,6 +824,7 @@ defmodule Ash.Test.Actions.LoadTest do
         |> Ash.Query.load(:latest_post)
         |> Api.read!()
 
+      refute author.latest_post.id == post1.id
       assert author.latest_post.id == post2.id
     end
   end