Merge branch 'main' of github.com:absinthe-graphql/absinthe into brya…

…njos/unbounded_concurrency_fix

.github/workflows/elixir.yml

@@ -4,7 +4,7 @@ on:
   push:
   pull_request:
     branches:
-      - master
+      - main
 
 jobs:
   test:

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 1.7.8
+
+- Bugfix: Fixes an issue where schemas would not find their types, or not be found at all.
+
 ## 1.7.7
 - POTENTIALLY BREAKING Bug Fix: [Validate variable usage in nested input arguments](https://github.com/absinthe-graphql/absinthe/pull/1290).This could break incoming documents previously considered valid. Skip the Absinthe.Phase.Document.Arguments.VariableTypesMatch phase to avoid this check. See Absinthe.Pipeline on adjusting the document pipeline.
 - #1321 resolves telemetry issues

lib/absinthe/subscription.ex

@@ -176,7 +176,7 @@ defmodule Absinthe.Subscription do
     registry = pubsub |> registry_name
 
     for field_key <- pdict_fields(doc_id) do
-      Registry.unregister(registry, field_key)
+      Registry.unregister_match(registry, field_key, doc_id)
     end
 
     Registry.unregister(registry, doc_id)
@@ -187,24 +187,22 @@ defmodule Absinthe.Subscription do
 
   @doc false
   def get(pubsub, key) do
-    pubsub
-    |> registry_name
-    |> Registry.lookup(key)
-    |> then(fn doc_ids ->
-      pubsub
-      |> registry_name
-      |> Registry.select(
-        # We compose a list of match specs that basically mean "lookup all keys
-        # in the doc_ids list"
-        for {_, doc_id} <- doc_ids,
-            do: {{:"$1", :_, :"$2"}, [{:==, :"$1", doc_id}], [{{:"$1", :"$2"}}]}
-      )
-    end)
-    |> Map.new(fn {doc_id, doc} ->
-      doc = Map.update!(doc, :initial_phases, &PipelineSerializer.unpack/1)
+    name = registry_name(pubsub)
 
-      {doc_id, doc}
+    name
+    |> Registry.lookup(key)
+    |> MapSet.new(fn {_pid, doc_id} -> doc_id end)
+    |> Enum.reduce([], fn doc_id, acc ->
+      case Registry.lookup(name, doc_id) do
+        [] ->
+          acc
+
+        [{_pid, doc} | _rest] ->
+          doc = Map.update!(doc, :initial_phases, &PipelineSerializer.unpack/1)
+          [{doc_id, doc} | acc]
+      end
     end)
+    |> Map.new()
   end
 
   @doc false

mix.exs

@@ -2,7 +2,7 @@ defmodule Absinthe.Mixfile do
   use Mix.Project
 
   @source_url "https://github.com/absinthe-graphql/absinthe"
-  @version "1.7.7"
+  @version "1.7.8"
 
   def project do
     [

test/absinthe/execution/subscription_test.exs

@@ -314,6 +314,39 @@ defmodule Absinthe.Execution.SubscriptionTest do
     refute_receive({:broadcast, _})
   end
 
+  test "can unsubscribe from duplicate subscriptions individually" do
+    client_id = "abc"
+
+    assert {:ok, %{"subscribed" => topic1}} =
+             run_subscription(
+               @query,
+               Schema,
+               variables: %{"clientId" => client_id},
+               context: %{pubsub: PubSub}
+             )
+
+    assert {:ok, %{"subscribed" => topic2}} =
+             run_subscription(
+               @query,
+               Schema,
+               variables: %{"clientId" => client_id},
+               context: %{pubsub: PubSub}
+             )
+
+    Absinthe.Subscription.publish(PubSub, "foo", thing: client_id)
+    assert_receive({:broadcast, a})
+    assert_receive({:broadcast, b})
+    doc_ids = Enum.map([a, b], & &1.topic)
+    assert topic1 in doc_ids
+    assert topic2 in doc_ids
+
+    Absinthe.Subscription.unsubscribe(PubSub, topic1)
+    Absinthe.Subscription.publish(PubSub, "bar", thing: client_id)
+    assert_receive({:broadcast, a})
+    refute_receive({:broadcast, _})
+    assert topic2 == a.topic
+  end
+
   @query """
   subscription {
     multipleTopics